In this post we are going to cover schemas and data contracts 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: five teams publish to shared topics, changes keep breaking consumers, and we introduce schemas safely.
A schema is a contract attached to a topic. It says what a message must look like, and Pub/Sub checks every message against it when it is published. Interviewers ask about this because broken events are one of the most common production problems in streaming. One team renames a field, and every consumer downstream breaks. A schema moves that failure to the team that made the change.
Part 8 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 publisher changed a field and consumers broke. How do Pub/Sub schemas help?
- Question 2: A teammate asks whether to use Avro or Protocol Buffers for a schema. How do you decide?
- Question 3: The schema must change without breaking existing consumers. How do schema revisions work?
- Question 4: Messages that don't match the schema reach the topic. What does schema validation reject?
- Question 5: Someone asks whether to use JSON or binary encoding. What are the trade-offs?
- Question 6: Your manager asks who owns an event contract between teams. How do you manage it?
- Scenario 7: Five teams publish to shared topics and changes keep breaking consumers. How do you introduce schemas safely?
- The full script
The short answer
A schema (Avro or Protocol Buffers) attached to a topic is checked on every publish: messages that don't match are not published. Revisions (up to 20 per schema) let the contract evolve, and a topic's revision range lets old and new publishers coexist. Validate payloads in CI before attaching a schema to a busy topic.
Let's set up the project
We run Google's official Pub/Sub emulator locally, with small Python scripts that use the google-cloud-pubsub client, so every output is real: Avro schemas, revisions, and validation on publish. The emulator doesn't support Protocol Buffer schemas, so those are shown as code, and revision IDs change on every run, so we never print them. The starting files hold the Avro schema order_event.avsc, its next version order_event_v2.avsc, and samples.py with the payloads we publish.
{
"type": "record",
"name": "OrderEvent",
"namespace": "shop",
"fields": [
{"name": "order_id", "type": "int"},
{"name": "status", "type": "string"},
{"name": "amount", "type": "double"}
]
}{
"type": "record",
"name": "OrderEvent",
"namespace": "shop",
"fields": [
{"name": "order_id", "type": "int"},
{"name": "status", "type": "string"},
{"name": "amount", "type": "double"},
{"name": "channel", "type": "string", "default": "web"}
]
}from google.cloud import pubsub_v1
from google.pubsub_v1.types import Encoding, Schema
project = "tb-shop-dev"
schemas = pubsub_v1.SchemaServiceClient()
path = schemas.schema_path(project, "order-event")
schema = schemas.create_schema(request={
"parent": f"projects/{project}", "schema_id": "order-event",
"schema": Schema(name=path, type_=Schema.Type.AVRO,
definition=open("order_event.avsc").read())})
print("schema", schema.name, schema.type_.name)
pub = pubsub_v1.PublisherClient()
for topic, enc in [("orders", Encoding.JSON),
("orders-binary", Encoding.BINARY)]:
t = pub.create_topic(request={
"name": pub.topic_path(project, topic),
"schema_settings": {"schema": path, "encoding": enc}})
print("topic", topic, "encoding", t.schema_settings.encoding.name)
sub = pubsub_v1.SubscriberClient()
reader = sub.subscription_path(project, "orders-reader")
sub.create_subscription(name=reader,
topic=pub.topic_path(project, "orders"))SAMPLES = {
"good": {"order_id": 1001, "status": "PAID", "amount": 25.5},
"renamed": {"order_id": 1002, "status": "PAID", "total": 40.0},
"wrong_type": {"order_id": "A-1003", "status": "PAID",
"amount": 9.9},
"null_status": {"order_id": 1004, "status": None, "amount": 12.0},
"not_json": b"order 1005 PAID 60.00",
"negative": {"order_id": 1006, "status": "PAID", "amount": -25.5},
"v1_order": {"order_id": 1007, "status": "PAID", "amount": 18.0},
"v2_order": {"order_id": 1008, "status": "PAID", "amount": 7.5,
"channel": "app"},
}import json, sys
from google.cloud import pubsub_v1
from samples import SAMPLES
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "orders")
for name in sys.argv[1:]:
data = SAMPLES[name]
if isinstance(data, dict):
data = json.dumps(data).encode()
try:
pub.publish(topic, data).result(timeout=30)
print(f"{name:12} published")
except Exception as e:
print(f"{name:12} rejected: {type(e).__name__} {e}")from google.cloud import pubsub_v1
from google.pubsub_v1.types import Schema
schemas = pubsub_v1.SchemaServiceClient()
path = schemas.schema_path("tb-shop-dev", "order-event")
rev = schemas.commit_schema(request={"name": path, "schema": Schema(
name=path, type_=Schema.Type.AVRO,
definition=open("order_event_v2.avsc").read())})
revs = list(schemas.list_schema_revisions(request={"name": path}))
print("new revision committed, id length:", len(rev.revision_id))
print("order-event now has", len(revs), "revisions")
topic = pubsub_v1.PublisherClient().get_topic(
topic="projects/tb-shop-dev/topics/orders")
s = topic.schema_settings
print("topic orders: first_revision_id", repr(s.first_revision_id),
"last_revision_id", repr(s.last_revision_id))import json
from google.cloud import pubsub_v1
project = "tb-shop-dev"
schemas = pubsub_v1.SchemaServiceClient()
revs = schemas.list_schema_revisions(
request={"name": schemas.schema_path(project, "order-event")})
label = {r.revision_id: "v2" if "channel" in r.definition else "v1"
for r in revs}
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path(project, "orders-reader")
rows = []
while True:
resp = sub.pull(subscription=path, max_messages=50, timeout=10)
if not resp.received_messages:
break
for m in resp.received_messages:
a = m.message.attributes
rows.append((json.loads(m.message.data)["order_id"],
a["googclient_schemaencoding"],
label[a["googclient_schemarevisionid"]]))
ids = [m.ack_id for m in resp.received_messages]
sub.acknowledge(subscription=path, ack_ids=ids)
for order_id, encoding, revision in sorted(rows):
print(f"order {order_id}: {encoding}, revision {revision}")import json, struct
from google.cloud import pubsub_v1
def zigzag(n): # Avro int and long: zigzag varint
n, out = (n << 1) ^ (n >> 63), b""
while n > 0x7F:
out += bytes([n & 0x7F | 0x80])
n >>= 7
return out + bytes([n])
def avro_binary(order_id, status, amount): # values in schema order
s = status.encode()
return (zigzag(order_id) + zigzag(len(s)) + s
+ struct.pack("<d", amount))
order = {"order_id": 1009, "status": "PAID", "amount": 25.5}
pub = pubsub_v1.PublisherClient()
for topic, data in [("orders", json.dumps(order).encode()),
("orders-binary", avro_binary(**order))]:
pub.publish(pub.topic_path("tb-shop-dev", topic), data).result()
print(f"{topic:13} {len(data):2} bytes {data!r}")TEAMS = { # one real payload from each publishing team
"checkout": {"order_id": 2001, "status": "PAID",
"amount": 30.0, "channel": "web"},
"mobile-app": {"order_id": "M-2002", "status": "PAID",
"amount": 12.5, "channel": "app"},
"partner-feed": {"order_id": 2003, "status": "PAID",
"total": 8.0, "channel": "partner"},
"returns": {"order_id": 2004, "status": "REFUNDED",
"amount": 30.0, "channel": "web"},
"loyalty": {"order_id": 2005, "status": None,
"amount": 0.0, "channel": "app"},
}import json, sys
from google.cloud import pubsub_v1
from google.pubsub_v1.types import Encoding
from team_samples import TEAMS
schemas = pubsub_v1.SchemaServiceClient()
name = schemas.schema_path("tb-shop-dev", "order-event")
failed = 0
for team, event in TEAMS.items():
try:
schemas.validate_message(request={
"parent": "projects/tb-shop-dev", "name": name,
"message": json.dumps(event).encode(),
"encoding": Encoding.JSON})
print(f"{team:13} PASS")
except Exception as e:
failed += 1
print(f"{team:13} FAIL {type(e).__name__} {e}")
print(failed, "of", len(TEAMS), "teams failed")
sys.exit(1 if failed else 0) # block the mergeQuestion 1: A publisher changed a field and consumers broke. How do Pub/Sub schemas help?
A schema is a contract attached to the topic, in Avro or Protocol Buffers. Pub/Sub checks every message when it is published, and a message that doesn't match is not published at all.
Our Avro schema for order events has three fields: order_id, status and amount. Let's create it, and two topics that use it: orders with JSON encoding, and orders-binary. We run python create_schema.py.
$ python create_schema.py
schema projects/tb-shop-dev/schemas/order-event AVRO
topic orders encoding JSON
topic orders-binary encoding BINARYWe get the schema order-event of type AVRO, topic orders encoding JSON and topic orders-binary encoding BINARY.
Now the publisher renames amount to total. We publish a good order and the renamed one with python publish.py good renamed.
$ python publish.py good renamed
good published
renamed rejected: InvalidArgument 400 Could not parse messageThe good order is published. The renamed one is rejected right at publish: InvalidArgument 400 Could not parse message.
That's the point. The error lands on the team that made the change, on their first publish, and consumers never see the broken event. Without a schema, every consumer finds out in production.
The publisher must check every publish future. A rejected message is not in the topic, so log it, alert, and keep the payload.
Question 2: A teammate asks whether to use Avro or Protocol Buffers for a schema. How do you decide?
Pub/Sub supports both: Avro 1.11, and Protocol Buffers with proto2 or proto3. Either way, a schema has one top-level type and no imports.
| Avro | Protocol Buffers | |
|---|---|---|
| Versions | Avro 1.11 | proto2 or proto3 |
| Definition | JSON (.avsc), one top-level record | .proto, one top-level message, no imports |
| Safe revision | Avro schema resolution, as reader and writer | add or remove optional fields only |
| Common fit | data teams, Avro files | services that already use gRPC |
| In this demo | runs on the emulator | code only: not in the emulator |
The real difference is how a schema may evolve. A Protocol Buffer revision may only add or remove optional fields. An Avro revision must follow Avro schema resolution, working both as a reader and as a writer of the other revisions.
So I pick what the teams already use. Services that talk gRPC already have proto files. Data teams that write Avro files, or define records in JSON, usually pick Avro.
Here is the same order event as a proto3 message. The emulator answers a Protocol Buffer schema with 501, not implemented, so this one is code only, and our demos use Avro.
syntax = "proto3";
package shop;
message OrderEvent {
int32 order_id = 1;
string status = 2;
double amount = 3;
optional string channel = 4; // added later
}Agree on the evolution rules up front. That is what keeps consumers safe, not the format.
Question 3: The schema must change without breaking existing consumers. How do schema revisions work?
We don't edit a schema, we commit a new revision. Version 2 adds channel, with a default of "web". A schema can have up to 20 revisions, and each gets an 8 character ID from the server. Let's commit it with python commit_v2.py.
$ python commit_v2.py
new revision committed, id length: 8
order-event now has 2 revisions
topic orders: first_revision_id '' last_revision_id ''It prints id length: 8 and order-event now has 2 revisions. The topic decides which revisions it accepts, with first_revision_id and last_revision_id. Both are empty here, '', so orders accepts all of them, and Pub/Sub tries them from newest to oldest.
Now old and new publishers both publish. We run python publish.py v1_order v2_order, then read the topic with python read.py. Pub/Sub adds attributes to each message, including googclient_schemaencoding and googclient_schemarevisionid, and our reader prints them.
$ python publish.py v1_order v2_order
v1_order published
v2_order published
$ python read.py
order 1001: JSON, revision v1
order 1007: JSON, revision v1
order 1008: JSON, revision v2Order 1001, the good order from question 1, and order 1007, the old publisher with no channel field, matched revision v1. Order 1008 matched revision v2.
During a migration, I set a revision range on the topic, so old and new publishers both work, and I tighten it when every publisher has moved.
gcloud pubsub topics update orders \
--schema=order-event --message-encoding=json \
--first-revision-id=FIRST_REVISION_ID \
--last-revision-id=LAST_REVISION_IDValidating a new definition doesn't check compatibility with older revisions, so test that in CI. A change can take a few minutes to take effect. And consumers should read the revision attribute.
Question 4: Messages that don't match the schema reach the topic. What does schema validation reject?
First, check that the topic really has the schema attached, and which revisions its range allows. A message is published only if its data parses against one of them, with the topic's encoding. Attributes are never checked.
Let's try four more payloads with python publish.py wrong_type null_status not_json negative.
$ python publish.py wrong_type null_status not_json negative
wrong_type rejected: InvalidArgument 400 Could not parse message
null_status rejected: InvalidArgument 400 Could not parse message
not_json rejected: InvalidArgument 400 Could not parse message
negative publishedThree are rejected with Could not parse message, and the negative amount is published. Here is why:
| Payload | Result | Why |
|---|---|---|
| "order_id": "A-1003" | rejected | string where the schema says int |
| "status": null | rejected | status is not nullable |
| order 1005 PAID 60.00 | rejected | not JSON |
| "amount": -25.5 | published | a valid double: business rules are not checked |
A schema checks structure and types, not business rules, so an amount of -25.5 still gets through.
The emulator's error text is Could not parse message. Don't match on the text. Treat a schema rejection as permanent: sending the same bytes again won't help.
A rejected payload goes to an error table or bucket with the reason, and business rules like positive amounts are checked by the producer or a consumer.
Question 5: Someone asks whether to use JSON or binary encoding. What are the trade-offs?
The encoding is set on the topic together with the schema: JSON or binary, and every message must use it. Let's publish the same order on both of our topics with python encoding.py.
$ python encoding.py
orders 52 bytes b'{"order_id": 1009, "status": "PAID", "amount": 25.5}'
orders-binary 15 bytes b'\xe2\x0f\x08PAID\x00\x00\x00\x00\x00\x809@'As JSON it is 52 bytes, and anyone can read it. As Avro binary it is 15 bytes: no field names, just the values in schema order. Smaller and faster to parse, but we need the schema to read it.
| JSON | Binary |
|---|---|
| readable | smaller |
| easy to debug | faster to parse |
| BigQuery table schema | needs the schema |
JSON wins for debugging, for partners, and for a BigQuery subscription that uses the table schema, which needs JSON messages. One Avro catch: in JSON, a union value, such as a nullable field that has a value, is wrapped with its type name.
Binary wins for high volume and for consumers that already use an Avro library. In production, use a library like fastavro, not hand-written bytes like this demo.
Changing a topic's encoding later can break active publishers.
Question 6: Your manager asks who owns an event contract between teams. How do you manage it?
The producing team owns it, because they own the data and they make the changes. But consumers get a vote: every change is a pull request that the consuming teams review.
Here is how it flows:
- Producer team - owns
order-event, opens the pull request - Git and the pipeline - review, validate, commit
- Consumer teams - review every change
The schema lives in Git next to the producer's code, and a pipeline commits new revisions. Nobody edits schemas by hand in the console.
Then lock it down with IAM. Pub/Sub has separate permissions to create, commit, roll back, delete, attach and validate schemas, so the pipeline can commit, the topic owner can attach, and CI can only validate.
| Who | Schema permissions |
|---|---|
| Release pipeline | pubsub.schemas.create, pubsub.schemas.commit, pubsub.schemas.rollback |
| Topic owner | pubsub.schemas.attach |
| CI checks | pubsub.schemas.validate |
| Platform admins only | pubsub.schemas.delete |
A schema can even live in a different project from the topic. I keep shared contracts in one platform project, so every team finds them in one place.
Which changes are allowed, how long an old revision stays in the range, and who to call when a publish starts failing.
Scenario 7: Five teams publish to shared topics and changes keep breaking consumers. How do you introduce schemas safely?
Don't switch validation on in one go. First, write the schema from the payloads the teams really send today, and let the producing teams agree on it.
Then test before we enforce. In CI, ci_check.py validates one real payload from each team against the schema, with validate_message. We run python ci_check.py.
$ python ci_check.py
checkout PASS
mobile-app FAIL InvalidArgument 400 Could not parse JSON Avro message
partner-feed FAIL InvalidArgument 400 Could not parse JSON Avro message
returns PASS
loyalty FAIL InvalidArgument 400 Could not parse JSON Avro message
3 of 5 teams failedCheckout and returns PASS. The mobile app sends a string order ID, the partner feed says total, and loyalty sends a null status, so all three FAIL with Could not parse JSON Avro message. The result is 3 of 5 teams failed, the script exits with an error, and nothing is enforced yet.
When every team passes, we attach the schema to the topic. The emulator can't update a topic, so this step is code only.
# attach the schema to an existing topic (not in the emulator)
publisher.update_topic(request={
"topic": {"name": topic_path, "schema_settings": {
"schema": schema_path, "encoding": "JSON"}},
"update_mask": {"paths": ["schema_settings"]}})Changing the schema or encoding on a live topic can break active publishers, so do one topic at a time. Here is the whole plan:
- Write the schema - from the payloads teams really send; producers agree
- Validate in CI - one real payload per team, before anything is enforced
- Attach to the topic - one topic at a time, when every team passes
- Evolve with revisions - a revision range during migrations, alert on publish errors
Surprises in production become failed builds, and that is what the business really wants.
Key points to remember
- A topic schema (Avro or Protocol Buffers) is checked on every publish
- Revisions: up to 20 per schema; the topic's revision range decides what is accepted
- Validation checks structure and types, not business rules
- JSON is readable, binary is smaller: choose once per topic
- Producers own the contract, consumers review, CI validates
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 #8 - Schemas and data contracts
# 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_08_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('order_event.avsc', """\
{
"type": "record",
"name": "OrderEvent",
"namespace": "shop",
"fields": [
{"name": "order_id", "type": "int"},
{"name": "status", "type": "string"},
{"name": "amount", "type": "double"}
]
}
""")
write('order_event_v2.avsc', """\
{
"type": "record",
"name": "OrderEvent",
"namespace": "shop",
"fields": [
{"name": "order_id", "type": "int"},
{"name": "status", "type": "string"},
{"name": "amount", "type": "double"},
{"name": "channel", "type": "string", "default": "web"}
]
}
""")
write('create_schema.py', """\
from google.cloud import pubsub_v1
from google.pubsub_v1.types import Encoding, Schema
project = "tb-shop-dev"
schemas = pubsub_v1.SchemaServiceClient()
path = schemas.schema_path(project, "order-event")
schema = schemas.create_schema(request={
"parent": f"projects/{project}", "schema_id": "order-event",
"schema": Schema(name=path, type_=Schema.Type.AVRO,
definition=open("order_event.avsc").read())})
print("schema", schema.name, schema.type_.name)
pub = pubsub_v1.PublisherClient()
for topic, enc in [("orders", Encoding.JSON),
("orders-binary", Encoding.BINARY)]:
t = pub.create_topic(request={
"name": pub.topic_path(project, topic),
"schema_settings": {"schema": path, "encoding": enc}})
print("topic", topic, "encoding", t.schema_settings.encoding.name)
sub = pubsub_v1.SubscriberClient()
reader = sub.subscription_path(project, "orders-reader")
sub.create_subscription(name=reader,
topic=pub.topic_path(project, "orders"))
""")
write('samples.py', """\
SAMPLES = {
"good": {"order_id": 1001, "status": "PAID", "amount": 25.5},
"renamed": {"order_id": 1002, "status": "PAID", "total": 40.0},
"wrong_type": {"order_id": "A-1003", "status": "PAID",
"amount": 9.9},
"null_status": {"order_id": 1004, "status": None, "amount": 12.0},
"not_json": b"order 1005 PAID 60.00",
"negative": {"order_id": 1006, "status": "PAID", "amount": -25.5},
"v1_order": {"order_id": 1007, "status": "PAID", "amount": 18.0},
"v2_order": {"order_id": 1008, "status": "PAID", "amount": 7.5,
"channel": "app"},
}
""")
write('publish.py', """\
import json, sys
from google.cloud import pubsub_v1
from samples import SAMPLES
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "orders")
for name in sys.argv[1:]:
data = SAMPLES[name]
if isinstance(data, dict):
data = json.dumps(data).encode()
try:
pub.publish(topic, data).result(timeout=30)
print(f"{name:12} published")
except Exception as e:
print(f"{name:12} rejected: {type(e).__name__} {e}")
""")
write('commit_v2.py', """\
from google.cloud import pubsub_v1
from google.pubsub_v1.types import Schema
schemas = pubsub_v1.SchemaServiceClient()
path = schemas.schema_path("tb-shop-dev", "order-event")
rev = schemas.commit_schema(request={"name": path, "schema": Schema(
name=path, type_=Schema.Type.AVRO,
definition=open("order_event_v2.avsc").read())})
revs = list(schemas.list_schema_revisions(request={"name": path}))
print("new revision committed, id length:", len(rev.revision_id))
print("order-event now has", len(revs), "revisions")
topic = pubsub_v1.PublisherClient().get_topic(
topic="projects/tb-shop-dev/topics/orders")
s = topic.schema_settings
print("topic orders: first_revision_id", repr(s.first_revision_id),
"last_revision_id", repr(s.last_revision_id))
""")
write('read.py', """\
import json
from google.cloud import pubsub_v1
project = "tb-shop-dev"
schemas = pubsub_v1.SchemaServiceClient()
revs = schemas.list_schema_revisions(
request={"name": schemas.schema_path(project, "order-event")})
label = {r.revision_id: "v2" if "channel" in r.definition else "v1"
for r in revs}
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path(project, "orders-reader")
rows = []
while True:
resp = sub.pull(subscription=path, max_messages=50, timeout=10)
if not resp.received_messages:
break
for m in resp.received_messages:
a = m.message.attributes
rows.append((json.loads(m.message.data)["order_id"],
a["googclient_schemaencoding"],
label[a["googclient_schemarevisionid"]]))
ids = [m.ack_id for m in resp.received_messages]
sub.acknowledge(subscription=path, ack_ids=ids)
for order_id, encoding, revision in sorted(rows):
print(f"order {order_id}: {encoding}, revision {revision}")
""")
write('encoding.py', """\
import json, struct
from google.cloud import pubsub_v1
def zigzag(n): # Avro int and long: zigzag varint
n, out = (n << 1) ^ (n >> 63), b""
while n > 0x7F:
out += bytes([n & 0x7F | 0x80])
n >>= 7
return out + bytes([n])
def avro_binary(order_id, status, amount): # values in schema order
s = status.encode()
return (zigzag(order_id) + zigzag(len(s)) + s
+ struct.pack("<d", amount))
order = {"order_id": 1009, "status": "PAID", "amount": 25.5}
pub = pubsub_v1.PublisherClient()
for topic, data in [("orders", json.dumps(order).encode()),
("orders-binary", avro_binary(**order))]:
pub.publish(pub.topic_path("tb-shop-dev", topic), data).result()
print(f"{topic:13} {len(data):2} bytes {data!r}")
""")
write('team_samples.py', """\
TEAMS = { # one real payload from each publishing team
"checkout": {"order_id": 2001, "status": "PAID",
"amount": 30.0, "channel": "web"},
"mobile-app": {"order_id": "M-2002", "status": "PAID",
"amount": 12.5, "channel": "app"},
"partner-feed": {"order_id": 2003, "status": "PAID",
"total": 8.0, "channel": "partner"},
"returns": {"order_id": 2004, "status": "REFUNDED",
"amount": 30.0, "channel": "web"},
"loyalty": {"order_id": 2005, "status": None,
"amount": 0.0, "channel": "app"},
}
""")
write('ci_check.py', """\
import json, sys
from google.cloud import pubsub_v1
from google.pubsub_v1.types import Encoding
from team_samples import TEAMS
schemas = pubsub_v1.SchemaServiceClient()
name = schemas.schema_path("tb-shop-dev", "order-event")
failed = 0
for team, event in TEAMS.items():
try:
schemas.validate_message(request={
"parent": "projects/tb-shop-dev", "name": name,
"message": json.dumps(event).encode(),
"encoding": Encoding.JSON})
print(f"{team:13} PASS")
except Exception as e:
failed += 1
print(f"{team:13} FAIL {type(e).__name__} {e}")
print(failed, "of", len(TEAMS), "teams failed")
sys.exit(1 if failed else 0) # block the merge
""")
# Step 1: An Avro schema, two topics that use it, and a reader subscription
run('python create_schema.py')
# Step 2: The publisher renames a field
run('python publish.py good renamed')
# Step 3: Commit revision v2: a new field with a default
run('python commit_v2.py')
# Step 4: Old and new publishers on the same topic
run('python publish.py v1_order v2_order')
run('python read.py')
# Step 5: What validation rejects, and what it lets through
run('python publish.py wrong_type null_status not_json negative')
# Step 6: The same order as JSON and as Avro binary
run('python encoding.py')
# Step 7: CI: validate every team's payload against the latest revision
run('python ci_check.py', expect_error=True) # EXPECTED TO FAIL: the video shows this error
Part 9: Pub/Sub to BigQuery and Dataflow
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.