Flow Control & Backpressure

Flow control trong Pub/Sub là cơ chế subscriber tự bảo vệ khỏi bị overwhelm bởi quá nhiều messages. Hiểu cơ chế này là điều kiện tiên quyết để giải quyết các vấn đề phổ biến nhất trong production: subscriber OOM, xử lý chậm làm backlog tăng không kiểm soát, hoặc ngược lại subscriber underutilized vì flow control quá conservative.

Tại sao cần flow control

Pub/Sub không biết subscriber đang xử lý bao nhiêu messages, cần bao nhiêu bộ nhớ, hay đang chậm vì lý do gì. Nếu không có flow control, Pub/Sub sẽ push (với StreamingPull) hoặc cho phép subscriber pull càng nhiều càng tốt — cho đến khi subscriber OOM, hoặc các messages đang xử lý quá nhiều dẫn đến timeout và cascade redelivery.

Hai kịch bản extreme không có flow control:

Kịch bản 1 — Subscriber overwhelm: Subscriber có thể nhận 10,000 messages/s từ Pub/Sub nhưng chỉ xử lý được 1,000 messages/s. Không có flow control, subscriber queue up 9,000 messages/s trong memory → OOM sau vài phút, hoặc ack deadline expire cho hàng nghìn messages → cascade redelivery → subscriber càng chậm hơn.

Kịch bản 2 — Resource waste: Subscriber có thể xử lý 1,000 messages/s nhưng flow control set quá thấp (100 outstanding). Throughput thực tế chỉ 100 messages/s dù có capacity cao hơn nhiều.

Dual-watermark model — maxOutstandingMessages và maxOutstandingBytes

High-level Pub/Sub client libraries implement flow control qua hai tham số:

maxOutstandingMessages (hay maxOutstandingElementCount)

Số lượng messages tối đa đang được hold bởi subscriber tại bất kỳ thời điểm nào — tức là messages đã được deliver từ Pub/Sub nhưng chưa được ack hoặc nack.

Khi số outstanding messages đạt threshold này, client library dừng pull thêm messages từ Pub/Sub (với StreamingPull: dừng request thêm; với Unary Pull: không gọi thêm pull requests). Subscriber có thể tiếp tục xử lý messages đã có trong local queue.

Default values (thay đổi theo library):

• Java: 1,000 messages
• Go: 1,000 messages
• Python: 1,000 messages
• C++: 1,000 messages

maxOutstandingBytes (hay maxOutstandingRequestBytes)

Tổng kích thước bytes của tất cả messages đang outstanding. Nếu tổng size vượt ngưỡng này, subscriber cũng dừng pull.

Default values:

• Java: 100 MiB
• Go: 100 MiB
• Python: không có default byte limit (chỉ message count)

Dual-watermark logic

Subscriber pause khi một trong hai threshold bị vượt, và resume khi cả hai đều drop xuống dưới threshold. Điều này có nghĩa:

Threshold: maxOutstandingMessages=1000, maxOutstandingBytes=50MB

State A: 999 messages, 49MB outstanding → tiếp tục pull
State B: 1001 messages, 10MB outstanding → PAUSE (message count exceeded)
State C: 500 messages, 51MB outstanding → PAUSE (byte size exceeded)
State D: 900 messages, 45MB outstanding → RESUME (cả hai dưới threshold)

Lý do dùng dual watermark: một message lớn (9MB) có thể chiếm nhiều bộ nhớ dù count chỉ là 1. Message count limit bảo vệ khỏi fan-out; byte limit bảo vệ khỏi large-message memory exhaustion.

Cấu hình flow control

// Java example
FlowControlSettings flowControlSettings = FlowControlSettings.newBuilder()
    .setMaxOutstandingElementCount(500L)        // tối đa 500 messages
    .setMaxOutstandingRequestBytes(50 * 1024 * 1024L)  // tối đa 50 MB
    .setLimitExceededBehavior(FlowController.LimitExceededBehavior.Block)
    .build();

Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver)
    .setFlowControlSettings(flowControlSettings)
    .build();

// Go example
client.Subscription("my-sub").ReceiveSettings = pubsub.ReceiveSettings{
    MaxOutstandingMessages: 500,
    MaxOutstandingBytes:    50 * 1024 * 1024,
    NumGoroutines:          4,
}

# Python example
flow_control = pubsub_v1.types.FlowControl(
    max_messages=500,
    max_bytes=50 * 1024 * 1024,
)
streaming_pull_future = subscriber.subscribe(
    subscription_path,
    callback=callback,
    flow_control=flow_control,
)

LimitExceededBehavior: Một số libraries có tùy chọn Block (block pull goroutine) hoặc ThrowException/Ignore. Block là default và behavior đúng cho flow control.

Flow control không giải quyết persistent overload

Đây là điểm quan trọng nhất cần hiểu: flow control là bộ đệm, không phải solution cho persistent overload.

Scenario minh họa

Pub/Sub backlog: 10,000,000 messages
Subscriber capacity: 1,000 msgs/s
Publisher rate: 2,000 msgs/s

Flow control: maxOutstandingMessages=1,000

Flow control giúp subscriber không bị OOM — chỉ giữ 1,000 messages trong memory tại một thời điểm. Nhưng:

• Publisher đang tạo 2,000 msgs/s
• Subscriber chỉ consume 1,000 msgs/s
• → Backlog tăng 1,000 msgs/s mãi mãi

Flow control không thể làm subscriber xử lý nhanh hơn. Nó chỉ prevent subscriber từ việc bị overwhelm momentarily. Solution cho persistent overload là scale out subscriber instances.

Theo docs: "If this scenario is a persistent state, rather than a transient spike in message volume, consider increasing the number of subscriber client instances."

Metrics để đưa ra quyết định scale

subscription/num_undelivered_messages

Số messages trong backlog của subscription chưa được deliver hoặc đã deliver nhưng chưa ack. Đây là primary metric để quyết định có cần scale subscriber không.

Trend analysis:
- Backlog tăng đều đặn → subscriber không đủ capacity, cần scale out
- Backlog ổn định dù có messages mới → subscriber đang kịp
- Backlog tăng đột ngột rồi giảm → transient spike, không cần scale

subscription/oldest_unacked_message_age

Tuổi của message unacked lâu nhất (tính bằng giây). Đây là signal quan trọng nhất về subscriber health:

• oldest_unacked_message_age tăng nhanh → subscriber stuck trên một message, hoặc một ordering key bị block
• oldest_unacked_message_age cao nhưng ổn định → subscriber đang xử lý chậm nhưng steady
• oldest_unacked_message_age gần bằng ack deadline → nguy hiểm: messages sắp expire và redeliver

Alert threshold thực tế: Set alert khi oldest_unacked_message_age > ack_deadline * 0.8. Nếu ack deadline là 60 giây, alert khi metric > 48 giây.

subscription/num_outstanding_messages

Số messages đang được held bởi subscribers (delivered nhưng chưa ack). Metric này phản ánh effective parallelism của subscriber fleet.

Nếu metric này bằng maxOutstandingMessages × số instances, flow control đang active — subscribers đang pull hết capacity. Nếu thấp hơn nhiều, subscriber đang idle hoặc Pub/Sub backlog trống.

Publisher-side metrics

• topic/send_message_operation_count: Rate publish của publishers
• topic/byte_cost: Bytes published per second

So sánh topic/send_message_operation_count với rate subscriber xử lý messages: nếu publish rate > consume rate, backlog sẽ tăng.

Subscriber scaling với backlog metrics

Autoscaling dựa trên backlog

KEDA (Kubernetes Event-Driven Autoscaling) tích hợp với Pub/Sub để scale subscriber pods dựa trên num_undelivered_messages:

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: pubsub-subscriber-scaler
spec:
  scaleTargetRef:
    name: my-subscriber-deployment
  triggers:
  - type: gcp-pubsub
    metadata:
      subscriptionName: "my-subscription"
      subscriptionSize: "100"      # scale khi backlog > 100 per pod
      mode: "SubscriptionSize"
      value: "100"
  minReplicaCount: 1
  maxReplicaCount: 50

subscriptionSize: "100" nghĩa là KEDA target mỗi pod xử lý 100 messages backlog. Nếu backlog là 1,000 → scale lên 10 pods.

Scale-to-zero với Pub/Sub

Khi backlog = 0, subscriber có thể scale xuống 0 pods (với KEDA). Khi có messages mới, KEDA detect backlog tăng và scale up. Đây là pattern cost-efficient cho workloads có burst pattern.

Cẩn thận với scale-to-zero và ordering keys: Nếu subscriber scale xuống 0 và scale up lại, state của ordering key delivery bị reset. Messages với ordering keys sẽ được redeliver từ đầu (vì ack deadline đã expire khi subscriber offline). Đảm bảo subscriber là idempotent.

Backpressure từ subscriber về Pub/Sub

Đây là một điểm tinh tế: flow control hoạt động theo chiều từ subscriber control việc nhận, không phải Pub/Sub control việc gửi.

Khi subscriber pause pull (vì flow control), Pub/Sub không "biết" subscriber đang overloaded. Pub/Sub chỉ thấy: subscription có backlog, nhưng không có subscriber nào pull. Điều này là đúng design — Pub/Sub không cần biết subscriber capacity. Subscriber tự manage.

Với push subscriptions, push window mechanism tạo ra backpressure ngầm: khi endpoint slow, push window không tăng (hoặc giảm), Pub/Sub tự nhiên gửi chậm lại. Đây là implicit backpressure dựa trên response time.

NumGoroutines/MaxConcurrentPull — parallelism trong một instance

Ngoài flow control messages count/bytes, subscriber còn có tham số kiểm soát parallelism bên trong một instance:

// Go: số goroutine xử lý messages concurrent
client.Subscription("my-sub").ReceiveSettings = pubsub.ReceiveSettings{
    NumGoroutines: 10,  // 10 goroutines xử lý song song
    ...
}

// Java: executor service với thread pool
Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver)
    .setParallelPullCount(4)  // 4 concurrent StreamingPull streams
    .setExecutorProvider(
        InstantiatingExecutorProvider.newBuilder()
            .setExecutorThreadCount(8)
            .build()
    )
    .build();

Relationship với flow control: NumGoroutines kiểm soát concurrency; flow control kiểm soát total outstanding. Nếu maxOutstandingMessages=1000 và NumGoroutines=10, trung bình mỗi goroutine xử lý 100 messages tại một thời điểm.

Đây không phải 10 kết nối riêng biệt — đây là 10 goroutines trên cùng một subscriber instance, cùng share flow control state.

Ví dụ: Tuning flow control cho batch processing

Use case: Subscriber process messages theo batch (gom 100 messages, ghi vào BigQuery một lần). Mỗi batch mất 2 giây.

Không có flow control tuning:

• Default: 1,000 outstanding messages
• Subscriber nhận 1,000 messages → batch 10 lần 100 → xử lý song song
• 10 goroutines × 2 giây = throughput 500 messages/s

Với tuning:

• maxOutstandingMessages=200 (2 batches đang xử lý cùng lúc)
• NumGoroutines=2 (2 batch đang chạy concurrent)
• Memory usage giảm 5x (không cần buffer 1,000 messages)
• Throughput: 200 messages / 2 giây = 100 messages/s (thấp hơn, nhưng có thể scale ra nhiều instances)

Việc tune flow control là tìm balance giữa throughput per instance và memory usage / resource consumption.

References

• Flow control
• Subscriber concurrency
• Monitoring subscriptions
• KEDA Pub/Sub scaler