TCP Flow Control and Kafka Poll: vì sao pull model vẫn có socket-level backpressure

Câu hỏi gốc

Kafka consumer là poll/pull model:

records = consumer.poll(Duration.ofMillis(100));

Nên mental model tự nhiên là:

Consumer muốn đọc lúc nào thì gọi poll(). Broker đâu có tự push message vô consumer. Vậy tại sao còn nói TCP flow control / receive buffer / consumer đọc không kịp?

Note này ghi lại điểm chưa rõ để research thêm. Cốt lõi là có nhiều tầng khác nhau:

Kafka application-level pull
  !=
TCP byte-stream delivery
  !=
Business logic processing records

Flow đơn giản hóa

Một Kafka FetchResponse đi qua nhiều tầng trước khi app thấy ConsumerRecords:

Kafka broker
  -> TCP send buffer
  -> network / NIC
  -> kernel TCP stack on consumer machine
  -> kernel socket receive buffer
  -> Kafka client read() / network I/O
  -> user-space ByteBuffer / client buffers
  -> parse FetchResponse
  -> verify checksum / decompress record batches
  -> ConsumerRecords returned by poll()
  -> application business logic processes records

Điểm dễ nhầm:

Business logic không xử lý trực tiếp từ TCP receive buffer. Nó xử lý records sau khi Kafka client đã đọc bytes khỏi socket và parse thành records.

Vậy câu “consumer đọc không kịp” nếu nói cho chuẩn phải là:

Consumer process / Kafka client không drain bytes từ kernel socket receive buffer sang user-space/client buffers đủ nhanh, hoặc không chạy network/poll loop đều, nên TCP receive buffer có thể đầy.

Không phải:

Business logic đọc từng Kafka message trực tiếp từ TCP buffer.

TCP flow control là gì?

TCP receiver có một receive buffer trong kernel. Receiver quảng cáo cho sender biết còn nhận được bao nhiêu bytes nữa qua receive window:

receive window = khoảng trống còn lại trong socket receive buffer

Nếu app/client đọc khỏi socket nhanh:

receive buffer được drain
-> window lớn
-> sender gửi tiếp thoải mái

Nếu app/client không đọc khỏi socket đủ nhanh:

receive buffer đầy
-> window nhỏ lại hoặc về 0
-> sender bị buộc gửi chậm/dừng

Flow control khác congestion control:

Sender thực tế bị giới hạn bởi cả hai:

allowed_send = min(receive_window, congestion_window)

Nhưng Kafka là poll mà?

Đúng. Kafka consumer chủ động fetch:

Consumer poll/fetch
  -> gửi FetchRequest
Broker
  -> trả FetchResponse

Pull model quyết định có request hay không, offset nào, partition nào, fetch size bao nhiêu.

Nhưng sau khi broker bắt đầu trả response, response đó vẫn là bytes chạy qua TCP:

FetchResponse 32MB
  -> TCP byte stream
  -> consumer socket receive buffer
  -> Kafka client read()

Một câu ngắn:

Kafka poll = xin hàng
TCP flow control = điều tiết vận chuyển hàng trong lúc hàng đang đi

Nếu không có pending fetch/response thì broker không tự gửi records mới. Nhưng với response đang truyền, TCP vẫn kiểm soát tốc độ theo khả năng receiver drain socket.

“Kernel báo app có data rồi” thì sao vẫn chậm?

I/O blocking hoặc event-driven thường có pattern:

1. Packet tới NIC
2. Kernel TCP stack nhận packet
3. Bytes vào socket receive buffer
4. Socket được mark readable
5. App/thread/event loop được wake hoặc được báo readable
6. App gọi read()
7. Bytes copy từ kernel buffer sang user-space buffer

“Readable” chỉ nghĩa là:

Có bytes trong kernel socket buffer để đọc.

Nó không nghĩa là:

• app đã đọc hết bytes;
• Kafka client đã parse xong response;
• record batches đã checksum/decompress xong;
• business logic đã xử lý xong records;
• app có CPU/scheduler time ngay lập tức.

App vẫn phải được OS scheduler cho chạy và gọi read() đủ nhanh. Nếu incoming rate cao hơn drain rate, receive buffer sẽ đầy:

incoming from broker: 100 MB/s
client read + parse:   40 MB/s
backlog growth:        60 MB/s

Socket buffer hữu hạn, nên cuối cùng receive window giảm.

Chỗ mental model dễ thiếu

Mental model ban đầu:

poll()
  -> all messages loaded into consumer RAM
  -> app processes them

Model này đúng ở mức app API: khi poll() trả về ConsumerRecords, records đó đã nằm trong memory consumer/client.

Nhưng thiếu các tầng trước poll() return:

kernel socket buffer
  -> Kafka client buffers
  -> protocol parsing
  -> record batch validation/decompression
  -> ConsumerRecords

Ngoài ra, poll() trong Kafka Java consumer không chỉ “lấy records đã có”. Nó còn drive nhiều việc của client:

poll()
  -> network I/O
  -> send fetches
  -> receive fetch responses
  -> heartbeat / rebalance handling
  -> parse completed responses
  -> return fetched records

Nếu app làm kiểu:

while (true) {
    ConsumerRecords<K, V> records = consumer.poll(timeout);
    process(records); // mất lâu
}

thì trong lúc process(records) chạy lâu, thread đó có thể không quay lại poll() để drive network I/O đều. Tùy client implementation/version/config, việc drain socket có thể bị ảnh hưởng.

Timeline ví dụ

t = 0s
consumer poll/fetch xin 32MB từ broker

 t = 0.1s
broker bắt đầu gửi FetchResponse qua TCP
kernel consumer nhận bytes vào socket receive buffer

 t = 0.2s
consumer app/JVM đang bận xử lý batch cũ, GC, parse/decompress, hoặc chưa quay lại poll loop
Kafka client không drain socket nhanh bằng tốc độ broker gửi

 t = 0.4s
socket receive buffer đầy dần
receive window nhỏ lại
broker gửi tiếp phần còn lại chậm hoặc tạm dừng

 t = 1.0s
consumer quay lại poll/read
socket buffer được drain
receive window mở lại
broker gửi tiếp

Từ ngoài nhìn có thể thấy:

broker healthy
consumer process alive
CPU tổng chưa chắc full
nhưng transfer bị stop-and-go

Checksum, packet order, record batch

TCP là byte stream. Packets ở network có thể bị split, merge, retransmit, hoặc đến lệch thứ tự, nhưng TCP stack expose cho application một stream ordered:

application read() sees ordered bytes

Kafka protocol nằm trên byte stream đó:

TCP ordered bytes
  -> Kafka protocol frame / FetchResponse
  -> record batches
  -> records by partition/offset

Kafka checksum/decompression thường xảy ra ở record batch layer sau khi bytes đã được read từ kernel socket buffer vào user-space/client buffer.

Vì vậy không nên nghĩ:

TCP packet -> Kafka message -> business logic

Mà nên nghĩ:

TCP byte stream -> Kafka response -> record batch -> records

Vậy khi nào TCP flow control đáng quan tâm?

Trong Kafka/Spark streaming, đa số case chậm nên debug tầng application trước:

- processing time / micro-batch duration
- sink latency
- consumer lag
- poll interval
- max.poll.records
- fetch.max.bytes / max.partition.fetch.bytes
- maxOffsetsPerTrigger trong Spark
- GC / memory pressure
- deserialization/decompression CPU

TCP receive window / zero window / retransmit / socket buffer metrics là tầng thấp hơn. Nó đáng nhìn khi nghi ngờ:

• large fetch responses;
• consumer bị GC pause hoặc CPU scheduling issue;
• network I/O thread/event loop không drain socket đều;
• receive buffer đầy/zero-window xuất hiện;
• broker send bị block dù application-level configs nhìn ổn.

Data engineering framing

Với Spark Structured Streaming đọc Kafka, backpressure chính thường ở tầng cao:

source rate
-> micro-batch processing
-> shuffle/state store
-> sink write
-> checkpoint commit

TCP flow control không thay thế Kafka poll/backpressure. Nó chỉ là tầng byte-stream dưới cùng, có thể làm FetchResponse truyền chậm nếu consumer side không drain socket/user-space pipeline kịp.

Nên nói chính xác hơn:

Kafka consumer control tốc độ ở tầng poll/fetch. TCP flow control chỉ điều tiết bytes của các fetch responses đang vận chuyển, dựa trên socket receive buffer của consumer.

Open questions để research thêm

• Trong Kafka Java Consumer hiện đại, chính xác network I/O được drive bởi app thread trong poll() tới mức nào? Có background thread nào drain socket riêng cho fetch response không, hay thread model khác nhau theo client?
• Spark Structured Streaming Kafka source dùng consumer pool/threading như thế nào? Thread nào gọi poll, thread nào process records?
• Khi app xử lý records lâu, socket receive buffer có thực sự đầy thường xuyên không, hay application-level max poll/fetch limits thường ngăn điều đó trước?
• Metrics nào xác nhận TCP flow control thật sự xảy ra? Ví dụ ss -tin, zero-window, receive queue, Kafka network metrics, broker request/send latency.
• Với TLS/SASL, encryption/decryption nằm ở đâu trong pipeline read/parse và ảnh hưởng drain rate thế nào?

Chốt mental model hiện tại

Kafka poll/pull:
consumer quyết định xin data khi nào và bao nhiêu.

TCP flow control:
receiver kernel quyết định sender được tiếp tục đẩy bytes nhanh cỡ nào trong lúc response đang truyền.

Business logic:
xử lý records sau khi Kafka client đã read/parse/decompress thành ConsumerRecords.

Câu cần nhớ:

“Đọc không kịp” trong TCP không có nghĩa business logic đọc message từ socket buffer. Nó nghĩa Kafka client/process không drain kernel socket buffer sang user-space/client pipeline đủ nhanh so với bytes đang tới.