Stream join & exactly-once — ghép dòng, đảm bảo không trùng

TL;DR: Stream join ghép hai topic thành kết quả mới; join hai stream bắt buộc có window để tránh state tích lũy vô hạn, còn join stream với KTable (bảng changelog) thì không cần. Exactly-once là thách thức lớn nhất trong stream — at-least-once mặc định tái gửi khi lỗi, gây đếm trùng. Kafka giải quyết bằng hai tầng: idempotent producer (producer ID + sequence number, dedup tại broker, từ Kafka 0.11.0, overhead ~0%) và transaction (atomic write nhiều partition + commit offset trong một transaction, overhead ~3% so với at-least-once). Pitfall kinh điển: bật enable.idempotence=true tưởng đủ exactly-once — nhưng chỉ bảo vệ phía ghi, toàn pipeline cần thêm transaction hoặc consumer idempotent.

Mỗi ngày một sàn thương mại điện tử xử lý triệu sự kiện từ hai dòng song song: orders (đơn đặt hàng) và payments (xác nhận thanh toán). Team muốn ghép hai stream này để ghi nhận "đơn thanh toán thành công" — nhưng gặp ngay hai bài toán: (1) payment và order đến không đồng bộ, lệch vài giây đến vài phút; (2) khi consumer crash rồi tái gửi, pipeline cộng doanh thu hai lần.

Bài này giải thích stream join (và vì sao ghép hai stream bắt buộc phải có cửa sổ thời gian), rồi đi vào exactly-once — hai tầng bảo vệ của Kafka để pipeline tính đúng dù lỗi tái gửi hay crash giữa chừng.

1. Analogy — hai băng chuyền song song

Hình dung hai băng chuyền chạy song song trong nhà kho: băng A mang kiện hàng từ kho lẻ, băng B mang phiếu thanh toán từ quầy thu ngân. Nhân viên ghép đơn cần khớp mỗi kiện với phiếu tương ứng — nhưng hai băng không đồng tốc, đôi khi kiện hàng tới trước phiếu vài phút.

Cách duy nhất để ghép: giữ kiện và phiếu trên bàn chờ trong một khoảng thời gian nhất định (ví dụ 10 phút). Không đặt giới hạn thời gian thì bàn chờ tích lũy mãi — nhà kho hết chỗ. Giới hạn quá ngắn thì phiếu tới trễ 11 phút bị bỏ sót. Đây chính xác là bài toán stream-stream windowed join.

2. Stream join — stream-stream windowed join

Trong Kafka Streams, join hai KStream (KStream-KStream) yêu cầu chỉ định một window. Cả hai bên được buffer trong cùng window; một event ở stream A khớp với event ở stream B nếu key bằng nhau và timestamp nằm trong khoảng window.

flowchart LR
  SA["Stream A: orders"] --> BUFA["Buffer A (10min)"]
  SB["Stream B: payments"] --> BUFB["Buffer B (10min)"]
  BUFA --> JOIN["Windowed Join<br/>(same key, same window)"]
  BUFB --> JOIN
  JOIN -->|"khop"| OUT["paid-orders"]
  JOIN -->|"het window, khong khop"| MISS["miss / late event"]

# Stream-stream windowed join
join(stream_orders, stream_payments, key=order_id, window=10min):

    for each order o at time t:
        payments <- buffer_B.get(key=o.order_id, range=[t-10min, t+10min])
        for each payment p in payments:
            emit(merge(o, p))         -- khop: ghi ra paid-orders

    -- symmetric: payment cung trigger check voi buffer A
    for each payment p at time t_prime:
        orders <- buffer_A.get(key=p.order_id, range=[t_prime-10min, t_prime+10min])
        for each order o in orders:
            emit(merge(o, p))

Kafka Streams tự quản lý state store cho buffer A và buffer B (dùng RocksDB). Mỗi bên giữ event chưa ghép trong khoảng window; event cũ hơn window tự động bị evict.

3. Vì sao stream-stream join bắt buộc phải có window?

Câu hỏi tưởng đơn giản: tại sao không join mà không có window — giữ toàn bộ lịch sử và khớp khi có dữ liệu mới?

Vấn đề là state không giới hạn. Mỗi event chưa khớp phải được giữ để chờ đối tác. Không có window, mọi event từ lúc khởi động pipeline đến nay đều cần lưu — state tăng tuyến tính theo thời gian chạy:

State voi window=10min (moi ben -> span ~20 phut): giu event trong [t-10min, t+10min]
    -- rate 1000 events/s moi stream: max state ~ 1.2 trieu event (van quan ly duoc)

State khong co window: giu tu luc bat dau
    -- sau 1 ngay: ~ 86.4 trieu events
    -- sau 1 tuan: ~ 604 trieu events --> OOM hoac disk full

Window giới hạn state: event cũ hơn window bị evict khỏi state store. Pipeline chạy mãi mà không rò rỉ bộ nhớ.

Đáp án:

A → stream-table join (KTable): user_profiles là bảng trạng thái — key là user_id, giá trị là profile mới nhất. Không cần window vì bạn muốn khớp click với profile hiện tại của user, không phải profile trong vài phút trước.

B → stream-stream windowed join: cả impressions và clicks đều là event unbounded. Cần window 5 phút để khớp impression với click xảy ra trong cùng khoảng thời gian.

4. Stream-table join — KTable changelog không cần window

KTable là abstraction của Kafka Streams biểu diễn bảng trạng thái: topic compacted, mỗi key giữ giá trị mới nhất. Join một KStream với KTable là non-windowed lookup — với mỗi event từ stream, tìm giá trị hiện tại của key trong KTable.

# Stream-table join (non-windowed)
join(stream_clicks, kTable_profiles, key=user_id):

    for each click c at time t:
        profile <- kTable.get(c.user_id)    -- lookup gia tri HIEN TAI, khong window
        if profile != null:
            emit(merge(c, profile))

GlobalKTable là biến thể: mỗi instance ứng dụng giữ bản sao đầy đủ của bảng, loại bỏ yêu cầu co-partition với stream. Dùng khi bảng nhỏ và cần join với stream ở bất kỳ partition nào.

5. Ba mức delivery — at-most-once, at-least-once, exactly-once là gì?

Khi producer gửi message và broker crash trước khi gửi ACK, producer không biết message đã được ghi hay chưa. Cách xử lý tình huống này tạo ra ba mức bảo đảm:

at-least-once là mặc định Kafka (acks=all). Nguồn trùng lặp chính trong pipeline hiện đại là vòng consume-process-produce: consumer đọc event, xử lý, ghi kết quả — rồi crash trước khi commit offset. Khi restart, nó đọc lại event đó từ offset cũ, xử lý và ghi kết quả lần hai → đếm trùng. (Trước Kafka 3.0, producer retry cũng có thể ghi lặp; từ 3.0 idempotent producer bật mặc định nên chặn lặp phía ghi trong cùng session — nhưng không chặn kịch bản reprocess ở trên.)

at-most-once (acks=0 hoặc không retry): gửi một phát, mất ACK thì mất luôn. Chấp nhận được cho log/metrics không quan trọng, không chấp nhận được cho billing.

Exactly-once phải giải quyết hai mặt: phía ghi (producer không tạo duplicate) và phía đọc–ghi–commit (consumer reprocess không cộng đôi). Kafka giải quyết bằng hai tầng riêng.

6. Exactly-once tầng 1 — idempotent producer

Từ Kafka 0.11.0 (2017), khi bật enable.idempotence=true, producer nhận một producer ID (PID) duy nhất từ broker khi khởi động. Mỗi message gửi đi kèm sequence number tăng đơn điệu theo từng partition.

sequenceDiagram
  participant P as Producer
  participant B as Broker
  P->>B: send(pid=42, seq=0, msg)
  B-->>P: ACK (mat tren mang)
  Note over P: Khong nhan ACK -- retry
  P->>B: retry: send(pid=42, seq=0, msg)
  B-->>P: ACK (seq=0 da co -- bo qua duplicate)
  Note over B: msg ghi dung 1 lan trong log

# Idempotent producer: dedup tai broker
producer.init():
    pid <- broker.allocatePID()    -- PID duy nhat cho session nay
    seq[partition] <- 0            -- sequence number theo partition

producer.send(msg, partition):
    msg.pid = pid
    msg.seq = seq[partition]
    seq[partition] <- seq[partition] + 1
    broker.append(msg)

# Broker logic khi nhan message:
if (pid, partition, seq) da co trong log:
    drop(msg)       -- dedup: bo qua, gui ACK lai
else:
    append(msg)     -- ghi moi, gui ACK

Overhead: ~0% — chỉ thêm vài byte int per message, không round-trip thêm.

Đáp án: Không. Idempotent producer chỉ dedup trong cùng một producer session (cùng PID). Khi process khởi động lại, PID mới được cấp — broker không biết message cũ là duplicate của session trước. Đây là giới hạn của idempotent producer: loại trùng do retry, không loại trùng do reprocess sau crash — đó là nhiệm vụ của transaction.

7. Exactly-once tầng 2 — Kafka transaction

Idempotent producer bảo vệ phía ghi (broker không nhận duplicate trong cùng session). Nhưng một pipeline stream điển hình cần làm ba việc nguyên tử: đọc từ input topic, ghi kết quả ra output topic, và commit offset của input. Nếu crash sau khi ghi output nhưng trước khi commit offset, consumer poll lại từ offset cũ và reprocess → ghi output lần hai.

Kafka transaction (từ 0.11.0) gói ba việc này vào một atomic transaction:

# Kafka transaction: atomic write + offset commit
producer.initTransactions(transactional_id="billing-eos-1")
    -- transactional_id: dinh danh ben vung qua cac lan restart

loop:
    records <- consumer.poll()
    producer.beginTransaction()

    for r in records:
        result <- process(r)
        producer.send("invoices", result)     -- ghi output

    producer.sendOffsetsToTransaction(consumer.position(), "billing-group")
    producer.commitTransaction()              -- atomic: ghi output + commit offset cung luc
    -- neu crash truoc commit: toan bo transaction abort
    -- consumer poll lai tu offset cu -> re-process -> cung ket qua

producer.initTransactions("order-eos-1")

for records in consumer.poll():
    producer.beginTransaction()
    for r in records:
        producer.send("order-summary", transform(r))
    producer.???         -- commit offset vao trong transaction
    producer.???         -- commit toan bo transaction

Đáp án:

    producer.sendOffsetsToTransaction(consumer.position(), "order-group")
    producer.commitTransaction()

Thứ tự quan trọng: sendOffsetsToTransaction gắn offset commit vào bên trong transaction đang mở — không commit offset riêng lẻ ngoài transaction. commitTransaction() ghi atomic cả hai.

Consumer phía đọc output cần bật isolation.level=read_committed để chỉ thấy message từ transaction đã commit — message từ transaction chưa hoàn tất hoặc đã abort không hiển thị.

Overhead: ~3% so với at-least-once (đo với commit 100ms, message 1KB). Overhead nhỏ nhất khi commit interval lớn (vài giây) và batch lớn — chi phí round-trip tới Transaction Coordinator được amortize qua nhiều message.

8. Pitfall tổng hợp

❌ SAI -- at-least-once tưởng là "ổn định":
   consumer crash -> poll lai tu offset cu -> process(r)
   -> send("invoices", result)  -- broker nhan them 1 ban sao
   -> billing dem 2 don hang

✅ ĐÚNG -- exactly-once qua transaction:
   producer.beginTransaction()
   send("invoices", result)
   sendOffsetsToTransaction(offset, group)
   commitTransaction()           -- abort -> re-process -> chi 1 ban sao

9. 📚 Deep Dive

10. Liên hệ các bài khác

• Bài 03 — Windowing: window trong stream join đến thẳng từ khái niệm windowing ở bài trước — tumbling/hopping/session window đều có thể làm join window; event-time + watermark quyết định khi nào window đóng và emit kết quả join.
• Bài 05 — Thiết kế stream pipeline: lab tổng hợp nơi bạn tự dựng pipeline ôm join + exactly-once và chọn window size, delivery guarantee phù hợp với SLA.
• Bài 02 — Log-based broker: idempotent producer và transaction xây trên cơ sở append-only log của Kafka — offset, partition, consumer group đều là nền cho cả hai cơ chế EOS.

11. Tóm tắt

• Stream-stream join bắt buộc phải có window để giới hạn state; không có window = buffer vô hạn = OOM theo thời gian chạy.
• Stream-table join (KStream-KTable) non-windowed: KTable biểu diễn giá trị mới nhất mỗi key (changelog compacted), lookup tại thời điểm event tới.
• Ba mức delivery: at-most-once (có thể mất), at-least-once (có thể trùng — mặc định Kafka), exactly-once (không mất, không trùng — cần cơ chế thêm).
• Idempotent producer (Kafka 0.11.0): producer ID + sequence number, broker dedup tự động, overhead ~0%. Chỉ bảo vệ phía ghi trong cùng session.
• Kafka transaction: atomic write nhiều partition + commit offset trong một transaction, overhead ~3%. Cần isolation.level=read_committed phía consumer đọc output.
• Pitfall lớn nhất: bật enable.idempotence=true tưởng đủ exactly-once — idempotent producer không bảo vệ reprocess sau crash. Exactly-once đầy đủ cần transaction.

12. Tự kiểm tra

Bài tiếp theo: Thiết kế stream pipeline — chọn kiến trúc đúng cho bài toán thực