Schemas & Consumer Scaling Patterns

Message schemas và consumer scaling patterns là hai chủ đề tách biệt nhưng cùng ảnh hưởng đến độ tin cậy và khả năng mở rộng của Pub/Sub pipeline. Schema enforce contract giữa publisher và subscriber — ngăn dữ liệu sai format vào hệ thống. Consumer scaling pattern xác định cách subscriber fleet được tổ chức để xử lý lượng messages đa dạng.

Message Schemas — Contract giữa Publisher và Subscriber

Tại sao cần schema

Pub/Sub message là opaque byte array — không có structure enforcement mặc định. Publisher có thể gửi bất cứ thứ gì, và subscriber phải parse. Trong distributed systems với nhiều teams:

• Team A (publisher) thay đổi JSON format nhưng không thông báo Team B (subscriber) → silent breakage
• Publisher gửi malformed data → subscriber crash hoặc skip messages
• Schema drift theo thời gian → technical debt, khó maintain

Schema giải quyết bằng cách enforce contract tại publish time: nếu message không conform schema, Pub/Sub từ chối publish và trả error cho publisher ngay lập tức. Subscriber không bao giờ nhận malformed messages (về mặt lý thuyết).

Avro và Protocol Buffers — hai supported formats

Apache Avro (version 1.11):

• Schema được viết bằng JSON
• Schema embedded trong serialized data (self-describing format tùy mode)
• Hỗ trợ complex types: records, arrays, maps, unions
• Good cho data pipelines, analytics workloads, schema registry patterns

Protocol Buffers (proto2 và proto3):

• Schema được viết bằng .proto syntax
• Binary format compact, không self-describing (cần schema để decode)
• Strong typing với backward/forward compatibility by design
• Phổ biến hơn trong microservices, gRPC ecosystems

Lựa chọn giữa Avro và Protobuf: Không có câu trả lời universal. Nếu team đang dùng gRPC/Protobuf, consistency cho phép dùng cùng .proto files cho cả Pub/Sub schemas. Nếu đang dùng Kafka/Avro schema registry pattern, Avro cho phép reuse tooling. Cả hai đều hỗ trợ schema evolution.

Schema validation và enforcement

Khi schema được gắn vào topic, Pub/Sub validate mỗi message khi publish:

Publisher gọi publish(message) → Pub/Sub deserialize với schema → 
  - Nếu valid: lưu vào storage, trả publish ACK
  - Nếu invalid: trả error "INVALID_ARGUMENT: schema validation failed", không lưu

Encoding specification: Khi gắn schema vào topic, cần chỉ định encoding:

• BINARY: message được serialized theo binary format của schema (compact, hiệu quả)
• JSON: message được encoding dưới dạng JSON string (human-readable, kém hiệu quả hơn)

Encoding là property của topic-schema binding, không phải schema. Cùng một schema có thể dùng với binary encoding trên topic production và JSON encoding trên topic development.

Schema object trong GCP

Schema là resource GCP độc lập, không gắn chặt vào topic:

# Tạo Avro schema
gcloud pubsub schemas create my-avro-schema \
  --type=AVRO \
  --definition='{
    "type": "record",
    "name": "Order",
    "fields": [
      {"name": "order_id", "type": "string"},
      {"name": "customer_id", "type": "string"},
      {"name": "amount", "type": "double"},
      {"name": "status", "type": {"type": "enum", "name": "Status", "symbols": ["PENDING", "CONFIRMED", "SHIPPED"]}}
    ]
  }'

# Gắn schema vào topic
gcloud pubsub topics create my-topic \
  --schema=my-avro-schema \
  --message-encoding=JSON

Resource limits:

• Tối đa 10,000 schemas per project
• Schema size tối đa: 300 KB
• Tối đa 20 revisions per schema

Schema evolution — revision model

Pub/Sub hỗ trợ schema evolution qua revision model: thay vì replace schema, bạn tạo revision mới. Topic có thể được cấu hình để chấp nhận messages từ một range of schema revisions.

# Tạo revision mới của schema (backward-compatible change)
gcloud pubsub schemas commit my-avro-schema \
  --definition='{
    "type": "record",
    "name": "Order",
    "fields": [
      {"name": "order_id", "type": "string"},
      {"name": "customer_id", "type": "string"},
      {"name": "amount", "type": "double"},
      {"name": "status", "type": {"type": "enum", "name": "Status", "symbols": ["PENDING", "CONFIRMED", "SHIPPED", "DELIVERED"]}},
      {"name": "metadata", "type": ["null", "string"], "default": null}  // Thêm optional field
    ]
  }'

Revision IDs: Mỗi revision được gán một revision ID tự động. Topic configuration có thể chỉ định:

• firstRevisionId: revision cũ nhất được chấp nhận
• lastRevisionId: revision mới nhất được chấp nhận

Điều này cho phép gradual rollout: publishers mới gửi với schema mới, publishers cũ vẫn gửi với schema cũ — cả hai đều được chấp nhận trong migration window.

Backward compatibility là trách nhiệm của bạn: Pub/Sub không enforce schema compatibility giữa revisions. Bạn có thể commit một revision incompatible, và messages từ publishers khác nhau (dùng schema versions khác nhau) sẽ coexist trong topic. Subscriber phải handle này.

Pub/Sub thêm attribute googclient_schemaencoding và googclient_schemarevisionid vào mỗi message — subscriber có thể dùng để route theo schema version.

Khi không cần schema

Schema validation overhead là không negligible — mỗi publish có thêm deserialization cost. Không phải mọi use case đều cần:

• Internal event bus với single publisher: schema drift ít rủi ro, overhead không worth
• Prototyping / development phase: schema thay đổi liên tục
• High-throughput telemetry với format cực kỳ đơn giản

Schema valuable nhất trong: multi-team systems, public/partner APIs, và production data pipelines nơi schema drift có business impact.

Consumer Scaling Patterns

Fan-out pattern — một topic, nhiều subscriptions

Đây là pattern cơ bản nhất và là lý do core của Pub/Sub tồn tại. Một message được publish một lần, nhưng nhiều downstream systems nhận được một copy độc lập:

Topic: "order-created"
  ├── Subscription "order-processor"    → Order processing service
  ├── Subscription "notification-svc"   → Email/SMS notification
  ├── Subscription "analytics-pipeline" → BigQuery analytics
  └── Subscription "fraud-detection"    → Fraud detection ML

Mỗi subscription hoạt động hoàn toàn độc lập:

• Backlog của mỗi subscription là riêng biệt
• Scale của từng subscriber không ảnh hưởng nhau
• Failure của một subscriber không block subscribers khác

Storage implication: Mỗi message phải được ack bởi tất cả subscriptions trước khi xóa. Nếu "analytics-pipeline" subscription bị lag (backlog lớn), message được giữ trong storage cho đến khi subscription đó ack. Đây là lý do fan-out có storage cost cao hơn single subscription.

Độc lập hoàn toàn: Fan-out trong Pub/Sub không phải "broadcast copies" theo nghĩa vật lý (tốn N lần storage). Pub/Sub dùng reference model — message data được lưu một lần, chỉ delivery state được nhân lên. Nhưng message không bị xóa cho đến khi tất cả subscriptions ack.

Competing consumers pattern — một subscription, nhiều subscriber instances

Đây là pattern để scale-out xử lý một consumer group. Nhiều instances cùng pull từ một subscription, mỗi instance nhận subset của messages:

Subscription: "order-processor"
  ├── Instance 1 (pod) → nhận messages [M1, M4, M7, ...]
  ├── Instance 2 (pod) → nhận messages [M2, M5, M8, ...]
  └── Instance 3 (pod) → nhận messages [M3, M6, M9, ...]

Không có partition assignment: Khác với Kafka consumer groups, Pub/Sub không assign partition cụ thể cho từng instance. Pub/Sub phân phối messages dynamically — instance nào pull thì nhận. Điều này đơn giản hơn (không cần rebalancing khi instance join/leave) nhưng cũng có limitations:

• Không guarantee cùng instance nhận messages liên quan
• Nếu cần affinity (cùng entity đến cùng instance), cần ordering keys

Scaling: Thêm instance → tăng tổng pull rate → giảm backlog. Số instances tối ưu phụ thuộc vào throughput của mỗi instance và publish rate.

Subscription-per-consumer-type vs shared subscription

Quyết định quan trọng khi thiết kế fan-out: nên dùng subscription riêng cho mỗi consumer type, hay một subscription shared cho nhiều types?

Subscription riêng (recommended):

Topic → Subscription A (consumer type A)
      → Subscription B (consumer type B)

• Mỗi consumer type có backlog riêng
• Failure/slow của A không ảnh hưởng B
• Monitor, alert, DLT per consumer type
• Rõ ràng về ownership và responsibility

Shared subscription với routing logic:

Topic → Subscription (tất cả consumers)
  → Consumer code tự phân loại và route

• Phức tạp hơn trong consumer code
• Backlog shared — một consumer slow ảnh hưởng tất cả
• Khó monitor per-consumer-type
• Không nên dùng trừ khi có lý do đặc biệt

Parallel subscription streams trong một process

Một subscriber process có thể open nhiều concurrent streams từ cùng một subscription. Điều này tăng throughput của một instance mà không cần thêm process:

# Python: flow control với max messages control parallelism
streaming_pull_future = subscriber.subscribe(
    subscription_path,
    callback=callback,
    flow_control=pubsub_v1.types.FlowControl(max_messages=500),
    # Num goroutines/threads handled by internal executor
)

// Java: explicit parallel pull count
Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver)
    .setParallelPullCount(4)  // 4 concurrent StreamingPull connections
    .build();

Tăng parallelism giúp tận dụng CPU, nhưng không phải silver bullet — nếu bottleneck là IO (database write, external API call), thêm goroutines không giúp nhiều.

Khi nào scale horizontal vs vertical

Scale vertical (tăng parallelism trong một instance):

• CPU hoặc I/O bound workloads
• Message processing là stateless
• Không cần ordering guarantee
• Đơn giản về deployment

Scale horizontal (thêm subscriber instances):

• Đã max parallelism trong một instance
• Muốn fault tolerance: nếu một instance down, instances khác tiếp tục
• Cần geographic distribution (xử lý gần user)
• Kubernetes/GKE với HPA hoặc KEDA autoscaling

KEDA autoscaling cho Pub/Sub (từ Chương 9) là pattern phổ biến nhất trong GKE:

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: order-processor-scaler
spec:
  scaleTargetRef:
    name: order-processor-deployment
  minReplicaCount: 2    # luôn có ít nhất 2 pod
  maxReplicaCount: 100
  triggers:
  - type: gcp-pubsub
    metadata:
      subscriptionName: "order-processor"
      subscriptionSize: "200"  # target 200 messages/pod

subscriptionSize: "200" nghĩa là KEDA calculate: desired_replicas = ceiling(backlog / 200). Nếu backlog = 1,000 → 5 pods. Nếu backlog = 0 → 2 pods (minReplicaCount).

References

• Message schemas
• Schema revisions
• Avro schema
• Protocol Buffers schema
• Subscriber best practices
• KEDA GCP Pub/Sub scaler