Flink 6 — Chịu lỗi & đảm bảo Exactly-once

Khi node lỗi, con số vẫn phải đúng

Trong một hệ streaming chạy 24/7, sự cố không phải là ngoại lệ mà là điều chắc chắn xảy ra: pod bị OOMKilled, TaskManager mất kết nối, node bị reboot khi vá bảo mật, mạng chập chờn vài giây. Câu hỏi không phải "có lỗi không" mà là "khi lỗi xảy ra, kết quả có còn đúng không".

Với pipeline ngân hàng, đây là câu hỏi sống còn. Một job đếm số giao dịch để tính hạn mức, một job cộng dồn tổng tiền chuyển trong ngày để cảnh báo rửa tiền, một job đối soát số dư — nếu sau một lần restart mà con số bị đếm thiếu (mất giao dịch) hoặc đếm thừa (tính trùng một giao dịch), thì kết quả sai lệch có thể dẫn tới chặn nhầm khách, bỏ lọt giao dịch đáng ngờ, hoặc lệch sổ. Không thể chấp nhận "gần đúng".

Bài này trả lời: Flink đảm bảo tính đúng đắn đến mức nào, bằng cơ chế gì, và bạn phải cấu hình những gì để đạt end-to-end exactly-once — không chỉ trong Flink mà cả chuỗi từ Kafka nguồn tới sink đích. Nó nối tiếp trực tiếp bài 4 về state & checkpoint: checkpoint là nền tảng, ở đây ta xem nó phối hợp với nguồn và sink ra sao để tạo ra bảo đảm toàn tuyến.

Ba mức delivery guarantee

"Delivery guarantee" là mức cam kết về việc mỗi sự kiện đầu vào ảnh hưởng đến kết quả bao nhiêu lần khi có sự cố. Có ba mức, từ yếu đến mạnh:

• at-most-once là mức "không làm gì cả": nếu lỗi trước khi xử lý xong, sự kiện đó biến mất. Đây thực chất là hệ không có cơ chế chịu lỗi. Chỉ chấp nhận được cho dữ liệu mà mất mát không gây hậu quả.
• at-least-once đảm bảo không mất, bằng cách phát lại (replay) từ nguồn sau khi lỗi. Nhưng vì phát lại, một số sự kiện đã xử lý trước lúc lỗi sẽ được xử lý lần nữa → kết quả có thể bị đếm trùng.
• exactly-once là mức mạnh nhất và là mặc định khuyến nghị cho pipeline tài chính.

Exactly-once là về EFFECT, không phải "xử lý vật lý một lần"

Đây là điểm gây hiểu lầm nhiều nhất, cần làm rõ dứt khoát. Exactly-once KHÔNG có nghĩa mỗi sự kiện được đưa qua code xử lý đúng một lần vật lý. Trên thực tế, khi Flink khôi phục từ checkpoint, nó buộc phải xử lý lại những sự kiện đã đi qua sau checkpoint đó — tức là chúng được xử lý vật lý nhiều lần.

Điều được đảm bảo "đúng một lần" là hiệu ứng (effect) lên trạng thái và kết quả cuối cùng: cho dù một sự kiện được chạy lại bao nhiêu lần trong quá trình khôi phục, tác động ròng của nó lên state và lên đầu ra chỉ được tính một lần duy nhất. Cách gọi chính xác hơn mà cộng đồng dùng là effectively exactly-once hay exactly-once state consistency.

Vì sao đạt được điều này? Vì khi restart, Flink:

1. Khôi phục state về đúng ảnh chụp (snapshot) tại checkpoint gần nhất — mọi thay đổi state sau checkpoint đó bị loại bỏ, coi như chưa từng xảy ra.
2. Tua nguồn (Kafka) về đúng offset đã ghi trong checkpoint đó và phát lại từ đấy.

Kết quả: state và vị trí đọc nguồn luôn khớp nhau như một khối nhất quán. Việc xử lý lại chỉ là để dựng lại đúng trạng thái mà checkpoint đã ghi nhận, không cộng thêm.

Ba điều kiện để đạt end-to-end exactly-once

Checkpoint chỉ đảm bảo đúng đắn bên trong Flink. Muốn đúng đắn toàn tuyến (end-to-end) — nghĩa là dữ liệu ghi ra hệ thống đích cũng không trùng/thiếu — cần đủ cả ba mắt xích:

Mắt xích 1 — Nguồn có thể replay (replayable source)

Sau khi lỗi, Flink phải tua nguồn về đúng vị trí của checkpoint và đọc lại. Nguồn phải cho phép điều đó. Kafka là ví dụ chuẩn: mỗi phân vùng có offset — con trỏ vị trí đọc — và Kafka giữ lại dữ liệu trong một khoảng thời gian (retention). Flink lưu offset của mọi phân vùng vào trong checkpoint, nên khi khôi phục nó biết đọc lại từ đâu.

Nguồn không tua lại được (ví dụ một socket TCP đọc là mất, hay HTTP endpoint không nhớ vị trí) thì không thể đạt exactly-once — cùng lắm là at-most-once, vì dữ liệu đã trôi qua không lấy lại được.

Mắt xích 2 — Checkpoint nhất quán trong Flink

Đây là cơ chế đã trình bày ở bài 4: Flink chèn checkpoint barrier vào luồng, các barrier trôi theo dữ liệu và kích hoạt mỗi toán tử chụp state của mình một cách nhất quán (thuật toán Chandy-Lamport). Kết quả là một ảnh chụp toàn cục khớp nhau: state của mọi toán tử + offset nguồn đều tương ứng cùng một "lát cắt" dữ liệu.

Mắt xích 3 — Sink hỗ trợ commit có điều kiện

Đây là mắt xích hay bị bỏ quên. State bên trong Flink có thể rollback, nhưng dữ liệu đã ghi ra ngoài thì không rút lại được. Nếu sink ghi ngay lập tức rồi Flink lỗi và replay, dữ liệu bị ghi trùng. Có hai cách để sink an toàn:

Cách A — Two-phase commit (2PC / transactional sink). Sink không lộ dữ liệu ngay mà theo hai pha khớp với vòng đời checkpoint:

• Pre-commit: khi barrier tới, sink ghi dữ liệu vào một giao dịch chưa commit (Kafka transaction, file tạm, JDBC transaction chưa commit).
• Commit: chỉ khi toàn bộ checkpoint hoàn tất thành công (mọi toán tử đã snapshot xong), JobManager mới báo sink commit giao dịch → lúc này dữ liệu mới thực sự hiện ra với hệ đích.
• Nếu lỗi giữa chừng, giao dịch chưa commit bị hủy → không có dữ liệu rác lộ ra.

Bên đọc phía sink Kafka phải đặt isolation.level=read_committed để không đọc dữ liệu của giao dịch chưa commit.

Cách B — Idempotent sink. Ghi trùng nhưng ghi đè thay vì cộng thêm: dùng khóa chính (primary key) và thao tác upsert (INSERT ... ON CONFLICT DO UPDATE, hoặc upsert của Elasticsearch/JDBC/Iceberg). Ghi lại cùng một bản ghi với cùng khóa nhiều lần cho kết quả giống hệt ghi một lần. Đơn giản hơn 2PC, không cần giao dịch, nhưng chỉ đúng khi thao tác thực sự idempotent (đè được, không phải cộng dồn ở phía đích).

Quy tắc vàng: exactly-once = min(guarantee của cả ba mắt xích). Chỉ cần một mắt xích ở mức at-least-once, toàn tuyến tụt xuống at-least-once.

Barrier alignment vs unaligned checkpoint

Khi một toán tử có nhiều luồng vào (ví dụ sau keyBy hoặc join, mỗi toán tử nhận từ nhiều toán tử thượng nguồn), barrier từ các luồng không tới cùng lúc. Cách xử lý điều này quyết định giữa exactly-once và độ trễ.

Aligned checkpoint (căn chỉnh barrier) — mặc định exactly-once

Toán tử chờ barrier từ MỌI luồng vào rồi mới snapshot. Luồng nào có barrier tới trước sẽ bị giữ lại (buffer) cho tới khi luồng chậm nhất cũng gửi barrier tới. Nhờ chờ căn chỉnh, snapshot phản ánh đúng một điểm cắt nhất quán → exactly-once.

Nhược điểm: nếu một luồng vào bị chậm (thường do backpressure), toán tử phải chờ, và thời gian căn chỉnh (alignment) có thể kéo dài → checkpoint lâu, thậm chí timeout. Đây chính là lý do dưới backpressure nặng, checkpoint hay thất bại.

Unaligned checkpoint — giảm độ trễ khi backpressure

Từ Flink 1.11, có unaligned checkpoint: toán tử không chờ căn chỉnh. Khi barrier đầu tiên tới, nó snapshot ngay, và đưa cả dữ liệu đang nằm trong buffer (những gì đã vượt qua barrier) vào snapshot. Barrier được cho "vượt" qua các buffer thay vì chờ.

Điểm quan trọng: unaligned vẫn giữ exactly-once, không hạ mức bảo đảm — nó chỉ đánh đổi bằng checkpoint lớn hơn và I/O nhiều hơn. Bật khi checkpoint liên tục timeout vì backpressure. Lưu ý unaligned không tương thích với một số tính năng (ví dụ khi kết hợp với at-least-once mode hoặc một vài loại state cũ).

// Minh hoạ cấu hình (không phải SQL, không đánh dấu chạy được)
CheckpointConfig cfg = env.getCheckpointConfig();
env.enableCheckpointing(30_000);                                  // checkpoint mỗi 30s
cfg.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);         // exactly-once
cfg.enableUnalignedCheckpoints();                                 // bật unaligned
cfg.setCheckpointTimeout(120_000);                                // 2 phút timeout
cfg.setMinPauseBetweenCheckpoints(10_000);                        // nghỉ 10s giữa 2 checkpoint
cfg.setTolerableCheckpointFailureNumber(3);                       // cho phép 3 lần fail liên tiếp

Khôi phục sau lỗi: restart strategy

Khi một task lỗi, Flink phục hồi bằng cách: (1) khôi phục state của toàn bộ job từ checkpoint hoàn tất gần nhất, (2) tua nguồn về offset trong checkpoint đó, (3) chạy lại. Nhưng chạy lại theo chiến lược nào — thử ngay hay chờ, thử bao nhiêu lần — do restart strategy quyết định. Cấu hình sai thì hoặc job "quay cuồng restart" đốt tài nguyên, hoặc bỏ cuộc quá sớm.

# flink-conf.yaml — minh hoạ exponential-delay (không đánh dấu chạy được)
restart-strategy: exponential-delay
restart-strategy.exponential-delay.initial-backoff: 10 s
restart-strategy.exponential-delay.max-backoff: 2 min
restart-strategy.exponential-delay.backoff-multiplier: 2.0
restart-strategy.exponential-delay.reset-backoff-threshold: 10 min

Với pipeline ngân hàng, exponential-delay hoặc failure-rate thường hợp lý hơn fixed-delay: khi Oracle/Kafka có sự cố, thử lại dồn dập chỉ làm nặng thêm; giãn dần cho hệ đích kịp hồi. Kết hợp với tolerableCheckpointFailureNumber để không fail job chỉ vì vài checkpoint lỡ nhịp.

Backpressure — khi hạ nguồn nghẽn, cả pipeline nghẹt

Backpressure (áp lực ngược) là hiện tượng khi một toán tử hạ nguồn (downstream) xử lý chậm hơn tốc độ dữ liệu đổ vào. Buffer của nó đầy, nó ngừng nhận thêm, khiến toán ttử thượng nguồn (upstream) gửi vào cũng bị chặn, buffer thượng nguồn đầy theo, và áp lực dồn ngược cho tới tận nguồn Kafka. Cuối cùng cả pipeline chạy chậm bằng khâu chậm nhất.

Backpressure không phải lỗi — nó là cơ chế tự vệ giúp Flink không bị tràn bộ nhớ (khác Spark cần điều chỉnh thủ công). Nhưng backpressure kéo dài là dấu hiệu bệnh: độ trễ tăng, và như đã nói, checkpoint alignment kéo dài dẫn tới checkpoint timeout — hai vấn đề đi cùng nhau.

Phát hiện backpressure

• Flink Web UI có tab Backpressure: mỗi toán tử hiển thị màu OK (xanh) / LOW / HIGH (đỏ). Toán ttử đỏ đầu tiên tính từ cuối lên thường không phải thủ phạm — thủ phạm là toán tử ngay sau nó (downstream) đang chậm.
• Các metric: isBackPressured, busyTimeMsPerSecond (toán tử bận bao nhiêu ms mỗi giây — gần 1000 là nghẽn), idleTimeMsPerSecond, độ trễ checkpoint và alignment time tăng.

Xử lý backpressure

1. Tìm đúng toán tử chậm: lần theo UI tới toán tử busy cao nhất ở cuối chuỗi — đó mới là nút thắt.
2. Tăng parallelism cho toán tử chậm để chia tải (chú ý key skew: một accountId nóng có thể vẫn dồn về một subtask).
3. Tối ưu operator chậm: sink JDBC ghi từng dòng → chuyển sang batch/async I/O; tra cứu bên ngoài đồng bộ → dùng Async I/O; hàm nặng → tối ưu thuật toán, giảm serialize.
4. Buffer debloating (từ Flink 1.14): taskmanager.network.memory.buffer-debloat.enabled — Flink tự động thu nhỏ lượng dữ liệu in-flight trong buffer mạng để rút ngắn thời gian alignment, giúp checkpoint kịp hoàn tất dưới backpressure.
5. Đích thực sự không theo kịp? Nâng cấu hình hệ đích, hoặc chèn Kafka làm vùng đệm giữa các tầng.

Watermark tương tác với checkpoint

Hai cơ chế này độc lập nhưng cùng chảy trong luồng và ảnh hưởng lẫn nhau:

• Watermark (xem bài 3) điều khiển thời gian sự kiện — quyết định khi nào window đóng và bắn kết quả. Watermark cũng là một phần của state và được lưu trong checkpoint, nên sau khôi phục, tiến độ thời gian sự kiện được khôi phục chính xác, window đang mở không bị mất.
• Checkpoint barrier điều khiển tính bền bỉ — quyết định khi nào snapshot state.

Điểm giao thoa quan trọng: dưới backpressure, cả watermark và checkpoint barrier đều bị kẹt lại vì cùng đi trong luồng dữ liệu bị nghẽn. Kết quả: window chậm đóng (kết quả ra trễ) và checkpoint chậm hoàn tất cùng lúc. Vì vậy khi thấy "kết quả window ra trễ" và "checkpoint timeout" xuất hiện đồng thời, thủ phạm chung thường là backpressure — sửa backpressure sẽ giải quyết cả hai.

Chi phí hiệu năng của exactly-once

Exactly-once không miễn phí. Cần cân nhắc thực tế:

• 2PC làm tăng độ trễ end-to-end: dữ liệu ở sink transactional chỉ hiện ra sau khi checkpoint hoàn tất. Nếu checkpoint mỗi 60s, độ trễ nhìn thấy dữ liệu ở đích có thể lên tới ~60s. Muốn dữ liệu tươi hơn thì checkpoint dày hơn — nhưng checkpoint quá dày lại tốn I/O và có thể chồng lấn.
• Chi phí giao dịch: Kafka transaction tốn overhead; số giao dịch mở đồng thời và thời gian giao dịch cần được giới hạn (transaction.timeout.ms phải ≥ chu kỳ checkpoint tối đa, nếu không giao dịch bị hủy giữa chừng → mất dữ liệu).
• Aligned checkpoint dễ timeout dưới backpressure → phải cân nhắc unaligned hoặc buffer debloating, đổi lấy checkpoint lớn hơn.

Kinh nghiệm: không phải job nào cũng cần exactly-once. Job audit/log thô có thể dùng at-least-once + dedup ở tầng sau để rẻ và nhanh hơn. Chỉ trả giá exactly-once cho những pipeline mà con số phải tuyệt đối đúng — đếm giao dịch, cộng dồn tiền, đối soát.

Use case thực tế

Bối cảnh. Team dữ liệu NCB xây pipeline đếm và cộng dồn giao dịch theo tài khoản trong ngày, phục vụ (1) cảnh báo giao dịch bất thường theo ngưỡng và (2) bảng đối soát cuối ngày. Nguồn là topic Kafka txn.events (CDC từ core banking qua Kafka), ~4.000 giao dịch/giây giờ cao điểm. Đích là bảng agg_txn_daily trên PostgreSQL và một topic Kafka txn.alerts.

Yêu cầu. Sau bất kỳ sự cố (TaskManager restart, node reboot), số đếm và tổng tiền của mỗi tài khoản không được thiếu hay trùng một giao dịch nào — vì lệch một giao dịch có thể chặn nhầm khách hoặc bỏ lọt cảnh báo.

Cấu hình đạt end-to-end exactly-once:

1. Nguồn: Kafka source, offset commit vào checkpoint (không dùng auto-commit của Kafka). Retention topic ≥ 7 ngày để luôn tua lại được.
2. Trong Flink: EXACTLY_ONCE, checkpoint mỗi 30s, RocksDB state backend (state lớn theo tài khoản), bật unaligned checkpoint vì giờ cao điểm có backpressure ở sink JDBC.
3. Sink JDBC (agg_txn_daily): dùng idempotent upsert theo khóa (account_id, business_date) — INSERT ... ON CONFLICT DO UPDATE. Replay ghi đè cùng khóa → an toàn, không cần 2PC.
4. Sink Kafka (txn.alerts): dùng 2PC transactional sink; consumer cảnh báo đặt isolation.level=read_committed.
5. Restart: exponential-delay (10s→2min) + tolerableCheckpointFailureNumber=3.

Sự cố thử nghiệm. Kill một TaskManager giữa giờ cao điểm. Flink phát hiện sau ~5s, restart toàn job, khôi phục state từ checkpoint gần nhất (cách thời điểm lỗi ~18s), tua Kafka về offset tương ứng, phát lại ~72.000 giao dịch của 18s đó. Vì upsert theo khóa và Kafka 2PC hủy giao dịch dở, kết quả cuối: số đếm và tổng tiền khớp chính xác với đường chạy không lỗi, txn.alerts không có cảnh báo trùng. Đối soát cuối ngày lệch 0 giao dịch.

Bài học vận hành. Ban đầu team để checkpoint 60s → độ trễ cảnh báo ~1 phút, nghiệp vụ than chậm. Giảm xuống 30s và bật buffer debloating: độ trễ cảnh báo giảm còn ~30s, checkpoint vẫn hoàn tất đều dưới backpressure. Trade-off exactly-once (độ trễ) được điều chỉnh bằng chu kỳ checkpoint, không phải hạ mức bảo đảm.

Ghi nhớ

• Ba mức: at-most-once (mất dữ liệu), at-least-once (không mất, có thể trùng), exactly-once (mỗi sự kiện ảnh hưởng kết quả đúng một lần) — mặc định cho pipeline tài chính.
• Exactly-once là về EFFECT (state/kết quả), KHÔNG phải xử lý vật lý một lần. Khi khôi phục, Flink có xử lý lại sự kiện; điều đảm bảo là tác động ròng lên state chỉ tính một lần. Tên chính xác: effectively exactly-once.
• End-to-end exactly-once cần đủ 3 mắt xích: nguồn replay được (Kafka offset trong checkpoint) + checkpoint nhất quán trong Flink + sink hỗ trợ (2PC hoặc idempotent upsert). Yếu một mắt xích → cả tuyến tụt xuống at-least-once.
• 2PC: pre-commit khi barrier, commit khi checkpoint hoàn tất; consumer sink Kafka đặt read_committed. Idempotent: upsert theo khóa, đơn giản hơn nhưng chỉ đúng khi ghi đè được.
• Aligned checkpoint: chờ căn chỉnh barrier, dễ timeout dưới backpressure. Unaligned checkpoint: không chờ, đưa in-flight data vào snapshot — vẫn exactly-once, đổi lấy checkpoint lớn hơn; bật khi backpressure kéo dài.
• Restart strategy: fixed-delay / exponential-delay / failure-rate / none. Ngân hàng nên dùng exponential hoặc failure-rate + tolerableCheckpointFailureNumber.
• Backpressure = downstream chậm → dồn ngược upstream tới tận nguồn. Phát hiện qua Web UI (tab Backpressure, metric busyTimeMsPerSecond); xử lý: tìm đúng operator chậm, tăng parallelism, tối ưu sink (batch/async I/O), buffer debloating.
• Watermark & checkpoint cùng chảy trong luồng: backpressure làm window chậm đóng và checkpoint chậm hoàn tất cùng lúc — sửa backpressure giải quyết cả hai.
• Chi phí exactly-once: 2PC tăng độ trễ (dữ liệu chỉ hiện sau checkpoint hoàn tất) — điều chỉnh bằng chu kỳ checkpoint. Không phải job nào cũng cần exactly-once; audit/log có thể at-least-once + dedup.