Pub/Sub Interview Questions and Answers Part 2 - Publishing and Message Attributes

Pub/Sub Interview Q&A · Part 2 of 10

In this post we are going to cover publishing and message attributes. 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 checkout service must publish every payment event without losing any during an outage, and you build the publisher.

Publishing looks like one line of code: call publish and move on. But that one call hides a lot. The client batches messages, retries errors, and holds pending messages in memory. Interviewers ask about it because most data loss and most slow pipelines start on the publisher side. If you can explain batching, attributes, retries, flow control and the size limits, you show that you have run a real publisher in production.

Part 2 of the Real Time Pub/Sub Interview Questions and Answers series on TechBrothersIT.

On this page

1. The short answer
2. Let's set up the project
3. Question 1: A teammate publishes one message at a time and throughput is low. How do batching settings help the publisher?
4. Question 2: Consumers need the event type without parsing the payload. How do message attributes work?
5. Question 3: A publish call fails during a spike. How do you handle publish errors and retries?
6. Question 4: The publisher runs out of memory when the topic is slow. What is publisher flow control?
7. Question 5: Someone asks what the message ID returned by publish means. How do you use it?
8. Question 6: Large files need to go through Pub/Sub. What are the message size limits, and what do you do instead?
9. Scenario 7: A checkout service must publish every payment event without losing any during an outage. How do you build the publisher?
10. The full script

The short answer

Say this first

The client batches messages (Python defaults: 100 messages, 1 MB, 10 ms), so collect the publish futures instead of waiting on each one. Put routing metadata in attributes, let the client retry transient errors, handle every future, and use flow control with BLOCK so spikes slow the publisher instead of exhausting memory. A message is at most 10 MB: send big files through Cloud Storage and publish a pointer.

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 clicks, payments and invoices topics in the project tb-shop-dev, and the payments-audit subscription on payments.

setup.py

from google.cloud import pubsub_v1

project = "tb-shop-dev"
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
for name in ["clicks", "payments", "invoices"]:
    publisher.create_topic(name=publisher.topic_path(project, name))
subscriber.create_subscription(
    name=subscriber.subscription_path(project, "payments-audit"),
    topic=publisher.topic_path(project, "payments"))

batching.py

import time
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1 import types

s = types.BatchSettings()                    # Python client defaults
print("defaults:", s.max_messages, "messages,", s.max_bytes,
      "bytes,", s.max_latency, "seconds")
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "clicks")

start = time.monotonic()
for _ in range(200):
    publisher.publish(topic, b"click").result()   # waits every time
one_by_one = time.monotonic() - start

start = time.monotonic()
futures = [publisher.publish(topic, b"click") for _ in range(200)]
ids = [f.result() for f in futures]               # waits once
batched = time.monotonic() - start
print("published 200 one by one, then", len(ids), "batched")
print("batched is more than 10 times faster:",
      one_by_one > 10 * batched)

attrs.py

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
events = [
    ("payment_captured", b'{"order_id": 1001, "amount": "59.90"}'),
    ("refund", b'{"order_id": 998, "amount": "15.00"}'),
]
for event_type, data in events:
    future = publisher.publish(topic, data, event_type=event_type,
                               source="checkout", schema_version="1")
    print(event_type, "published as message", future.result())

route.py

from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()
path = subscriber.subscription_path("tb-shop-dev", "payments-audit")
resp = subscriber.pull(subscription=path, max_messages=10, timeout=10)
msgs = sorted(resp.received_messages,
              key=lambda r: r.message.message_id)
for r in msgs:
    attrs = dict(sorted(r.message.attributes.items()))
    if attrs["event_type"] == "refund":        # payload not parsed
        handler = "refund handler"
    else:
        handler = "payment handler"
    print(r.message.message_id, attrs, "->", handler)
subscriber.acknowledge(subscription=path,
                       ack_ids=[r.ack_id for r in msgs])

safe_publish.py

import sys
from google.api_core import exceptions, retry
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", sys.argv[1])
short = retry.Retry(initial=0.2, maximum=2.0, multiplier=2.0,
                    timeout=5.0)      # demo only: default 600 s
future = publisher.publish(topic, b'{"payment_id": "pay-7001"}',
                           retry=short)
try:
    print("published as message", future.result())
except exceptions.NotFound as e:
    print("not retried:", e)
except exceptions.RetryError as e:
    print("gave up:", e.message + ", last error:",
          type(e.cause).__name__)

flow.py

import sys
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1 import types

mode = sys.argv[1]                          # BLOCK, ERROR or IGNORE
flow = types.PublishFlowControl(
    message_limit=5, byte_limit=1000,
    limit_exceeded_behavior=types.LimitExceededBehavior[mode])
publisher = pubsub_v1.PublisherClient(
    batch_settings=types.BatchSettings(max_latency=0.5),  # slow topic
    publisher_options=types.PublisherOptions(flow_control=flow))
topic = publisher.topic_path("tb-shop-dev", "clicks")
futures = [publisher.publish(topic, b"click") for _ in range(50)]
ok, failed, error = 0, 0, None
for f in futures:
    try:
        f.result()
        ok += 1
    except Exception as e:
        failed, error = failed + 1, error or e
print(mode, "published", ok, "failed", failed)
if error:
    print(type(error).__name__ + ":", str(error).split(" - ")[0])

ids.py

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
sub = subscriber.subscription_path("tb-shop-dev", "payments-audit")

data = b'{"payment_id": "pay-7002", "amount": "40.00"}'
for attempt in ["first send:", "sent again:"]:
    future = publisher.publish(topic, data, payment_id="pay-7002")
    print(attempt, "message", future.result())

def pull():
    return subscriber.pull(subscription=sub, max_messages=10,
                           timeout=10).received_messages

first = pull()
print("delivered:", sorted(r.message.message_id for r in first))
subscriber.modify_ack_deadline(                 # a nack
    subscription=sub, ack_deadline_seconds=0,
    ack_ids=[r.ack_id for r in first])
again = pull()
print("redelivered:", sorted(r.message.message_id for r in again))
subscriber.acknowledge(subscription=sub,
                       ack_ids=[r.ack_id for r in again])

big.py

from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.publisher import exceptions

publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "invoices")
for size in [9_000_000, 10_485_760]:          # 9 MB, then 10 MiB
    try:
        msg_id = publisher.publish(topic, b"x" * size).result()
        print(f"{size:,} bytes: published as message {msg_id}")
    except exceptions.MessageTooLargeError as e:
        print(f"{size:,} bytes: MessageTooLargeError")
        print(e)

checkout.py

import json
from google.api_core import retry
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1 import types

flow = types.PublishFlowControl(
    message_limit=1000, byte_limit=10_000_000,
    limit_exceeded_behavior=types.LimitExceededBehavior.BLOCK)
publisher = pubsub_v1.PublisherClient(
    publisher_options=types.PublisherOptions(flow_control=flow))
topic = publisher.topic_path("tb-shop-dev", "payments")
short = retry.Retry(initial=0.2, maximum=2.0, timeout=5.0)  # demo
futures = {}
for pid in ["pay-8001", "pay-8002", "pay-8003"]:
    data = json.dumps({"payment_id": pid}).encode()
    futures[pid] = publisher.publish(topic, data, payment_id=pid,
                                     retry=short)
saved = 0
with open("outbox.jsonl", "a") as outbox:
    for pid, future in futures.items():
        try:
            future.result()
        except Exception:                     # retries are used up
            outbox.write(json.dumps({"payment_id": pid}) + "\n")
            saved += 1
print(f"published: {len(futures) - saved}, saved to outbox: {saved}")

replay_outbox.py

import json, os
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
with open("outbox.jsonl") as f:
    rows = [json.loads(line) for line in f]
for row in rows:
    pid = row["payment_id"]
    future = publisher.publish(topic, json.dumps(row).encode(),
                               payment_id=pid)
    print(pid, "published as message", future.result())
os.remove("outbox.jsonl")         # after every publish worked

read_payments.py

from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()
path = subscriber.subscription_path("tb-shop-dev", "payments-audit")
resp = subscriber.pull(subscription=path, max_messages=10, timeout=10)
msgs = resp.received_messages
ids = sorted(r.message.attributes["payment_id"] for r in msgs)
print("payments-audit received", len(ids), "payments:", ids)
subscriber.acknowledge(subscription=path,
                       ack_ids=[r.ack_id for r in msgs])

Question 1: A teammate publishes one message at a time and throughput is low. How do batching settings help the publisher?

The client libraries batch for you. A publish call doesn't send right away: it adds the message to a batch, and the batch goes out when it reaches a message count, a byte size or a time limit, whichever comes first.

The Python client's defaults are 100 messages, 1 MB, or 10 ms. The classic mistake is calling result after every publish. Each call waits for the batch timer and a round trip, so every request carries one message.

Let's try 200 clicks both ways with python batching.py. First it waits on every publish, then it collects all the futures and waits once at the end.

Terminal

$ python batching.py
defaults: 100 messages, 1000000 bytes, 0.01 seconds
published 200 one by one, then 200 batched
batched is more than 10 times faster: True

The defaults print as 100 messages, 1000000 bytes, 0.01 seconds. Then published 200 one by one, then 200 batched, and batched is more than 10 times faster: True.

Batching also cuts cost. Every publish request is billed for at least 1 KB, so tiny messages sent one per request pay for more than they send. One request can carry up to 1,000 messages and 10 MB.

Tuning

Raise max latency and max messages for high volume streams, and keep them low when every millisecond counts. Batching always trades a little latency for throughput.

Question 2: Consumers need the event type without parsing the payload. How do message attributes work?

Attributes are key-value strings that travel next to the data. Put routing metadata there, like event type, source and schema version, and keep the business payload in the data.

In Python, attributes are keyword arguments to publish. Let's run python attrs.py. It publishes a captured payment and a refund to the payments topic, each with three attributes.

Terminal

$ python attrs.py
payment_captured published as message 401
refund published as message 402

payment_captured published as message 401, and refund published as message 402. Now the consumer, python route.py, reads only the attributes to pick a handler.

Terminal

$ python route.py
401 {'event_type': 'payment_captured', 'schema_version': '1', 'source': 'checkout'} -> payment handler
402 {'event_type': 'refund', 'schema_version': '1', 'source': 'checkout'} -> refund handler

Message 401 has 'event_type': 'payment_captured' and goes -> payment handler. Message 402 has 'event_type': 'refund' and goes -> refund handler. The consumer never parsed the payload.

Attributes also power subscription filters, which work on attributes only, so a refunds team can get just the refunds. Here are the limits on the real service.

Attribute limit (real service)	Value
attributes per message	up to 100
key size	up to 256 bytes
value size	up to 1,024 bytes

Two pitfalls

Values must be strings, so send a number as text. And the emulator doesn't enforce those size limits, so don't learn them from local tests.

Question 3: A publish call fails during a spike. How do you handle publish errors and retries?

The client already retries. The Python client's default publish retry starts at 0.1 seconds, backs off by a factor of 4 up to 60 seconds between tries, and gives up after 600 seconds in total (without message ordering). It retries errors like Unavailable, DeadlineExceeded and ResourceExhausted.

So your job is the future. Let's try two failures with safe_publish.py. For the demo it uses a short retry with a 5 second timeout. First a topic name with a typo, then an outage: the client points at localhost:1, where nothing listens.

Terminal

$ python safe_publish.py payments-typo
not retried: 404 Topic not found
$ PUBSUB_EMULATOR_HOST=localhost:1 python safe_publish.py payments
gave up: Timeout of 5.0s exceeded, last error: ServiceUnavailable

The typo gives not retried: 404 Topic not found, and it comes back at once. That's a config bug. Fix it, don't loop on it. The outage gives gave up: Timeout of 5.0s exceeded, last error: ServiceUnavailable. Unavailable was retried until the timeout, and then the future failed with a retry error.

In production:

• Log every failure - with the business key
• Retry transient errors - the client does it, with backoff
• Save what needs a person - permission denied: store it durably, replay later

Pitfall

A retry can publish a message twice, when the first attempt worked but the response was lost. The two copies get different message IDs, so put a business key like payment_id in the message for consumers to dedupe.

Question 4: The publisher runs out of memory when the topic is slow. What is publisher flow control?

Publisher flow control caps how many messages and bytes wait in the client until the server confirms them. Without it, when publishing is slow, pending messages pile up in memory until they fail with deadline exceeded.

You set a message limit, a byte limit, and the behavior at the limit.

limit_exceeded_behavior	At the limit
BLOCK	publish() waits: back pressure
ERROR	the future fails with FlowControlLimitError
IGNORE (Python default)	nothing enforced (default limits: 1,000 messages, 10 MB)

The Python default is 1,000 messages and 10 MB with IGNORE, so nothing is enforced.

Let's see the difference. In flow.py the limit is 5 messages, and each batch is held for half a second, standing in for a slow topic. We publish 50 at once, first with python flow.py ERROR, then with python flow.py BLOCK.

Terminal

$ python flow.py ERROR
ERROR published 5 failed 45
FlowControlLimitError: Flow control limits would be exceeded
$ python flow.py BLOCK
BLOCK published 50 failed 0

With ERROR, ERROR published 5 failed 45, and the futures fail with FlowControlLimitError: Flow control limits would be exceeded. With BLOCK, publish just waits, and we get BLOCK published 50 failed 0.

So for data you can't lose, use BLOCK: back pressure slows the producer instead of dropping messages. Use ERROR only when the caller can shed load, like optional click events.

Real project

Size the limits from the memory you can spare. If publish blocks a lot, check publish latency and errors before you raise them.

Question 5: Someone asks what the message ID returned by publish means. How do you use it?

The message ID is assigned by the server when a publish succeeds, and it's unique within the topic. The publish future returns it.

Let's run python ids.py. It publishes the same payment twice, the way a publisher retry would. Then it pulls both, nacks them, and pulls again.

Terminal

$ python ids.py
first send: message 458
sent again: message 459
delivered: ['458', '459']
redelivered: ['458', '459']

The first send is message 458, and the second is message 459. Same payment, two message IDs, so to Pub/Sub those are two messages. On the subscriber side it's the other way round. Both are delivered: ['458', '459'], and after the nack they're redelivered: ['458', '459']. A redelivery keeps the ID.

Duplicate from	Message ID	Dedupe on
redelivery (nack, expired deadline)	the same	message ID or business key
publisher retry	a new one	business key (payment_id)

So use it for tracing. Log the ID when you publish and when you consume, and you can show whether a message was lost or delivered twice. Deduping on the message ID catches redeliveries, but not publisher retries, so dedupe on a business key like the payment ID.

Caution

The emulator hands out small counting numbers. In production, treat message IDs as opaque strings, never as a sequence, and store them with the topic name.

Question 6: Large files need to go through Pub/Sub. What are the message size limits, and what do you do instead?

A message can be up to 10 MB, and one publish request up to 10 MB and 1,000 messages. The Python client checks the request size before it sends anything.

Limit (real service)	Value
message data	up to 10 MB
one publish request	up to 10 MB and 1,000 messages
Python client check, before sending	10,000,000 bytes per request

Let's try both sides of the limit with python big.py: 9 MB, then 10 MiB.

Terminal

$ python big.py
9,000,000 bytes: published as message 460
10,485,760 bytes: MessageTooLargeError
The message being published would produce too large a publish request that would exceed the maximum
  allowed size on the backend (10000000 bytes).

9,000,000 bytes: published as message 460. But 10,485,760 bytes: MessageTooLargeError, because it would exceed the maximum allowed size on the backend (10000000 bytes). Nothing reached the server.

Even under the limit, big messages hurt. Consumers hold them in memory, and every subscription pays delivery throughput for every byte.

So for large files, use the claim check pattern. Upload the file to Cloud Storage, and publish a small message with its gs:// path and size. Cloud Storage isn't part of the emulator, so this part is shown as code.

claim_check.py

# claim check: the file goes to Cloud Storage, Pub/Sub gets a pointer
import json
from google.cloud import storage     # publisher, topic as before

bucket = storage.Client().bucket("tb-shop-invoices")
blob = bucket.blob("2026/10/inv-1001.pdf")
blob.upload_from_filename("inv-1001.pdf")       # the big file
ref = {"uri": f"gs://{bucket.name}/{blob.name}", "bytes": blob.size}
publisher.publish(topic, json.dumps(ref).encode(),
                  event_type="invoice_ready").result()  # pointer

Real project

Publish only after the upload has finished, so the object is there when the consumer looks. Still handle a missing object, and give the bucket its own cleanup rules.

Scenario 7: A checkout service must publish every payment event without losing any during an outage. How do you build the publisher?

Start with the basics:

• Handle every future - success or failure, never fire and forget
• Client retries with backoff - keep the defaults for transient errors
• Flow control: BLOCK - a spike slows checkout, nothing is dropped

But retries end. Let's simulate an outage: we run checkout.py with the client pointed at localhost:1, so it can't reach Pub/Sub, and the retry timeout runs out, 5 seconds for the demo. Then we look at the outbox file with cat outbox.jsonl.

Terminal

$ PUBSUB_EMULATOR_HOST=localhost:1 python checkout.py
published: 0, saved to outbox: 3
$ cat outbox.jsonl
{"payment_id": "pay-8001"}
{"payment_id": "pay-8002"}
{"payment_id": "pay-8003"}

It prints published: 0, saved to outbox: 3, and the outbox holds pay-8001, pay-8002 and pay-8003. No payment is lost.

Now Pub/Sub is reachable again. We run python replay_outbox.py, and then python read_payments.py to check the audit subscription.

Terminal

$ python replay_outbox.py
pay-8001 published as message 461
pay-8002 published as message 462
pay-8003 published as message 463
$ python read_payments.py
payments-audit received 3 payments: ['pay-8001', 'pay-8002', 'pay-8003']

The replay publishes the three payments as messages 461, 462 and 463, and payments-audit received 3 payments: ['pay-8001', 'pay-8002', 'pay-8003'].

In a real project, the outbox is a table written in the same database transaction as the payment, and a relay publishes from it and then marks the rows. That's the transactional outbox pattern, and it also survives a crash of the checkout service itself.

Every message carries its payment_id as an attribute, so consumers can dedupe after a retry or a replay.

Alert

Alert when outbox rows stay unpublished for more than a few minutes.

Key points to remember

• The client batches: collect futures, don't wait on each publish
• Attributes: string metadata for routing and filters
• The client retries transient errors; you handle every future
• Flow control with BLOCK keeps memory bounded during spikes
• 10 MB per message: big files go through Cloud Storage

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

Full script - writes the files and runs every step against the Pub/Sub emulator (pip install google-cloud-pubsub), run it with Python

# ========================================================================
# Pub/Sub Interview Q&A #2 - Publishing and message attributes
# 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_02_Code.py   in an empty folder.
# Steps that restart the emulator say so in a comment: stop and start it by hand at that point.
# ========================================================================

import os, subprocess, time

def write(path, text):
    if text is None:
        if os.path.exists(path): os.remove(path)
        return
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(text.strip("\n") + "\n")

def run(cmd, expect_error=False, allow_error=False, env=None):
    print("\n$ " + cmd, flush=True)
    rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
    if rc != 0 and not (expect_error or allow_error):
        raise SystemExit("stopped: this command was expected to succeed")

# Step 0: the starting files
write('setup.py', """\
from google.cloud import pubsub_v1

project = "tb-shop-dev"
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
for name in ["clicks", "payments", "invoices"]:
    publisher.create_topic(name=publisher.topic_path(project, name))
subscriber.create_subscription(
    name=subscriber.subscription_path(project, "payments-audit"),
    topic=publisher.topic_path(project, "payments"))
""")
write('batching.py', """\
import time
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1 import types

s = types.BatchSettings()                    # Python client defaults
print("defaults:", s.max_messages, "messages,", s.max_bytes,
      "bytes,", s.max_latency, "seconds")
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "clicks")

start = time.monotonic()
for _ in range(200):
    publisher.publish(topic, b"click").result()   # waits every time
one_by_one = time.monotonic() - start

start = time.monotonic()
futures = [publisher.publish(topic, b"click") for _ in range(200)]
ids = [f.result() for f in futures]               # waits once
batched = time.monotonic() - start
print("published 200 one by one, then", len(ids), "batched")
print("batched is more than 10 times faster:",
      one_by_one > 10 * batched)
""")
write('attrs.py', """\
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
events = [
    ("payment_captured", b'{"order_id": 1001, "amount": "59.90"}'),
    ("refund", b'{"order_id": 998, "amount": "15.00"}'),
]
for event_type, data in events:
    future = publisher.publish(topic, data, event_type=event_type,
                               source="checkout", schema_version="1")
    print(event_type, "published as message", future.result())
""")
write('route.py', """\
from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()
path = subscriber.subscription_path("tb-shop-dev", "payments-audit")
resp = subscriber.pull(subscription=path, max_messages=10, timeout=10)
msgs = sorted(resp.received_messages,
              key=lambda r: r.message.message_id)
for r in msgs:
    attrs = dict(sorted(r.message.attributes.items()))
    if attrs["event_type"] == "refund":        # payload not parsed
        handler = "refund handler"
    else:
        handler = "payment handler"
    print(r.message.message_id, attrs, "->", handler)
subscriber.acknowledge(subscription=path,
                       ack_ids=[r.ack_id for r in msgs])
""")
write('safe_publish.py', """\
import sys
from google.api_core import exceptions, retry
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", sys.argv[1])
short = retry.Retry(initial=0.2, maximum=2.0, multiplier=2.0,
                    timeout=5.0)      # demo only: default 600 s
future = publisher.publish(topic, b'{"payment_id": "pay-7001"}',
                           retry=short)
try:
    print("published as message", future.result())
except exceptions.NotFound as e:
    print("not retried:", e)
except exceptions.RetryError as e:
    print("gave up:", e.message + ", last error:",
          type(e.cause).__name__)
""")
write('flow.py', """\
import sys
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1 import types

mode = sys.argv[1]                          # BLOCK, ERROR or IGNORE
flow = types.PublishFlowControl(
    message_limit=5, byte_limit=1000,
    limit_exceeded_behavior=types.LimitExceededBehavior[mode])
publisher = pubsub_v1.PublisherClient(
    batch_settings=types.BatchSettings(max_latency=0.5),  # slow topic
    publisher_options=types.PublisherOptions(flow_control=flow))
topic = publisher.topic_path("tb-shop-dev", "clicks")
futures = [publisher.publish(topic, b"click") for _ in range(50)]
ok, failed, error = 0, 0, None
for f in futures:
    try:
        f.result()
        ok += 1
    except Exception as e:
        failed, error = failed + 1, error or e
print(mode, "published", ok, "failed", failed)
if error:
    print(type(error).__name__ + ":", str(error).split(" - ")[0])
""")
write('ids.py', """\
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
sub = subscriber.subscription_path("tb-shop-dev", "payments-audit")

data = b'{"payment_id": "pay-7002", "amount": "40.00"}'
for attempt in ["first send:", "sent again:"]:
    future = publisher.publish(topic, data, payment_id="pay-7002")
    print(attempt, "message", future.result())

def pull():
    return subscriber.pull(subscription=sub, max_messages=10,
                           timeout=10).received_messages

first = pull()
print("delivered:", sorted(r.message.message_id for r in first))
subscriber.modify_ack_deadline(                 # a nack
    subscription=sub, ack_deadline_seconds=0,
    ack_ids=[r.ack_id for r in first])
again = pull()
print("redelivered:", sorted(r.message.message_id for r in again))
subscriber.acknowledge(subscription=sub,
                       ack_ids=[r.ack_id for r in again])
""")
write('big.py', """\
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.publisher import exceptions

publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "invoices")
for size in [9_000_000, 10_485_760]:          # 9 MB, then 10 MiB
    try:
        msg_id = publisher.publish(topic, b"x" * size).result()
        print(f"{size:,} bytes: published as message {msg_id}")
    except exceptions.MessageTooLargeError as e:
        print(f"{size:,} bytes: MessageTooLargeError")
        print(e)
""")
write('checkout.py', 'import json\nfrom google.api_core import retry\nfrom google.cloud import pubsub_v1\nfrom google.cloud.pubsub_v1 import types\n\nflow = types.PublishFlowControl(\n    message_limit=1000, byte_limit=10_000_000,\n    limit_exceeded_behavior=types.LimitExceededBehavior.BLOCK)\npublisher = pubsub_v1.PublisherClient(\n    publisher_options=types.PublisherOptions(flow_control=flow))\ntopic = publisher.topic_path("tb-shop-dev", "payments")\nshort = retry.Retry(initial=0.2, maximum=2.0, timeout=5.0)  # demo\nfutures = {}\nfor pid in ["pay-8001", "pay-8002", "pay-8003"]:\n    data = json.dumps({"payment_id": pid}).encode()\n    futures[pid] = publisher.publish(topic, data, payment_id=pid,\n                                     retry=short)\nsaved = 0\nwith open("outbox.jsonl", "a") as outbox:\n    for pid, future in futures.items():\n        try:\n            future.result()\n        except Exception:                     # retries are used up\n            outbox.write(json.dumps({"payment_id": pid}) + "\\n")\n            saved += 1\nprint(f"published: {len(futures) - saved}, saved to outbox: {saved}")\n')
write('replay_outbox.py', """\
import json, os
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
with open("outbox.jsonl") as f:
    rows = [json.loads(line) for line in f]
for row in rows:
    pid = row["payment_id"]
    future = publisher.publish(topic, json.dumps(row).encode(),
                               payment_id=pid)
    print(pid, "published as message", future.result())
os.remove("outbox.jsonl")         # after every publish worked
""")
write('read_payments.py', """\
from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()
path = subscriber.subscription_path("tb-shop-dev", "payments-audit")
resp = subscriber.pull(subscription=path, max_messages=10, timeout=10)
msgs = resp.received_messages
ids = sorted(r.message.attributes["payment_id"] for r in msgs)
print("payments-audit received", len(ids), "payments:", ids)
subscriber.acknowledge(subscription=path,
                       ack_ids=[r.ack_id for r in msgs])
""")
run('python setup.py')

# Step 1: Batching: 200 clicks one by one, then batched
run('python batching.py')

# Step 2: Publish two payment events with attributes
run('python attrs.py')

# Step 3: Route on attributes without parsing the payload
run('python route.py')

# Step 4: A typo in the topic name, then an outage (nothing listens on localhost:1)
run('python safe_publish.py payments-typo')
run('PUBSUB_EMULATOR_HOST=localhost:1 python safe_publish.py payments')

# Step 5: Publisher flow control: ERROR versus BLOCK
run('python flow.py ERROR')
run('python flow.py BLOCK')

# Step 6: Message IDs: the same payment published twice, then a nack
run('python ids.py')

# Step 7: Message size: 9 MB, then 10 MiB
run('python big.py')

# Step 8: Scenario: an outage during checkout, payments go to the outbox
run('PUBSUB_EMULATOR_HOST=localhost:1 python checkout.py')
run('cat outbox.jsonl')

# Step 9: Scenario: Pub/Sub is back, replay the outbox and check the audit subscription
run('python replay_outbox.py')
run('python read_payments.py')

Next in this series

Part 3: Subscriptions: Pull and Push

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.