Exactly-once không tự có: Kafka, Flink và ClickHouse

"Exactly-once" là một cụm từ rất dễ viết vào tài liệu thiết kế. Flink có checkpoint exactly-once, Kafka có producer idempotent, ClickHouse có bảng tự khử trùng. Ghép ba thứ đó lại, cảm giác như dữ liệu đã an toàn.

Bài này đi qua một pipeline phân tích sự kiện mình từng làm dở: webhook từ GitHub, Jira đi vào một gateway, qua Kafka, được Flink tổng hợp, rồi ghi xuống ClickHouse. Khi đọc lại code để viết bài, mình thấy ở mỗi chặng đều có một chỗ dữ liệu có thể mất hoặc bị đếm trùng, dù tài liệu thiết kế nói ngược lại. Mỗi chỗ là một bài học khá phổ biến, nên mình ghi lại cùng cách làm đúng.

Ba chặng có thể làm mất hoặc đếm trùng dữ liệu, dù công cụ nào cũng quảng cáo exactly-once.

Một nguyên tắc dùng xuyên suốt: exactly-once đầu-cuối gần như luôn là at-least-once cộng với một đích ghi idempotent. Câu hỏi ở mỗi chặng là: nếu bước này chạy lại, kết quả có giữ nguyên không?

Chặng 1: gateway nhận khoá trước khi việc xong

Nhà cung cấp webhook gửi lại khi không nhận được phản hồi thành công, nên gateway cần khử trùng. Cách làm ban đầu:

if (!idempotencyStore.claim(event.idempotencyKey())) {   // Redis SET NX
    return duplicate();
}
publisher.publish(event);   // gửi Kafka bất đồng bộ, không chờ kết quả
return accepted();          // trả 2xx ngay

Vấn đề nằm ở thứ tự. Khoá được giữ trước khi sự kiện thật sự vào Kafka, và gateway trả 2xx mà không chờ Kafka xác nhận. Nếu Kafka lỗi đúng lúc đó:

Nhà cung cấp đã nhận 2xx, nên không gửi lại.
Nếu có gửi lại, lần đó bị coi là trùng, vì khoá đã nằm trong Redis.

Lần gửi lại từ phía nhà cung cấp, cách phục hồi rẻ nhất, bị lãng phí. Đường lui còn lại là dead letter queue (DLQ), nhưng DLQ thường cũng nằm trên chính cụm Kafka đang lỗi.

Cách làm đúng, tuỳ mức chặt chẽ cần có:

Chờ Kafka xác nhận rồi mới trả 2xx. Đây là thay đổi quan trọng nhất. Gửi lỗi thì trả 5xx để nhà cung cấp gửi lại.
Chỉ đánh dấu "đã xử lý" sau khi gửi thành công, hoặc dùng hai trạng thái: pending khi bắt đầu (có TTL ngắn), done khi Kafka xác nhận, và xoá khoá khi gửi lỗi.
Nếu phía sau đã có một lớp khử trùng chính xác (ở đây là Flink, xem dưới), lớp ở gateway chỉ cần là tối ưu, không cần là bảo đảm. Hai lần gửi song song cùng lọt qua thì cũng không sao, vì Flink sẽ bỏ bản thứ hai.

Bài học chung cho mọi kiểu khoá idempotency: khoá chỉ được nói "đã xong" khi tác dụng phụ đã xong. Giữ khoá trước tác dụng phụ là biến lỗi tạm thời thành mất dữ liệu vĩnh viễn.

Chặng 2: Flink khử trùng bằng keyed state

Ở Flink, sự kiện được keyBy theo orgId:source:eventId, và một ValueState ghi nhớ khoá đã thấy:

Long firstSeen = firstSeenState.value();
if (firstSeen == null) {
    firstSeenState.update(event.eventTime().toEpochMilli());
    out.collect(event);
}
// có rồi: bỏ qua, tăng bộ đếm dedup_dropped_total

Chặng này làm đúng. State nằm trong checkpoint, nên sau khi job khởi động lại từ checkpoint, cả offset Kafka lẫn tập khoá đã thấy đều quay về cùng một thời điểm. State có TTL để không phình mãi, và TTL chỉ gia hạn khi ghi, không gia hạn khi đọc (OnCreateAndWrite): một bản trùng đến sau 23 giờ vẫn bị bắt, và việc kiểm tra không làm khoá sống lâu hơn.

Điều cần nhớ là thời hạn khử trùng phải phủ được khoảng cách lớn nhất giữa hai lần gửi trùng (ví dụ thời gian nhà cung cấp còn thử gửi lại), chứ không liên quan tới thời gian Kafka giữ dữ liệu. Vài chi tiết dễ hiểu sai của state TTL nằm ở bài thời gian trong Apache Flink.

Chặng 3: sink có bộ đệm nhưng không gắn với checkpoint

Ghi ClickHouse từng dòng một thì quá chậm, nên sink gom thành lô 1.000 dòng. Phiên bản đầu kế thừa RichSinkFunction, giữ một List làm bộ đệm, và có comment:

// Hết lượt retry: ghi log và bỏ batch. Checkpoint sẽ không tiến,
// nên khi job khởi động lại batch này sẽ được ghi lại.

Comment này sai, và là kiểu sai rất dễ mắc:

Checkpoint của Flink chụp state của các operator. Một List bình thường trong sink không phải state, và sink không làm gì khi checkpoint xảy ra. Barrier đi qua sink, checkpoint hoàn tất, offset Kafka tiến lên, trong khi các dòng đó vẫn chỉ nằm trong RAM.
Nếu job chết lúc này, khởi động lại từ checkpoint sẽ đọc tiếp sau các dòng đó. Chúng mất hẳn.
Batch bị "bỏ" sau khi retry hết lượt cũng vậy: không có gì ngăn checkpoint tiến, nên batch đó không bao giờ được ghi lại.

Thêm một chi tiết: comment nói lô được đẩy đi mỗi 5 giây "bằng một luồng nền", nhưng code chỉ kiểm tra thời gian khi có bản ghi mới đến. Stream im lặng thì dữ liệu nằm trong bộ đệm vô thời hạn.

Cách làm đúng:

public class ClickHouseSink extends RichSinkFunction<Row> implements CheckpointedFunction {
    @Override
    public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
        flush();   // đẩy hết bộ đệm trước khi checkpoint được phép hoàn tất
    }

    private void flush() throws Exception {
        // retry có giới hạn; hết lượt thì NÉM lỗi chứ không bỏ batch,
        // để checkpoint thất bại và job khởi động lại từ checkpoint trước
    }
}

Flush trong snapshotState (hàm của CheckpointedFunction): checkpoint chỉ hoàn tất khi mọi dòng trước barrier đã nằm trong ClickHouse. Đây là điều kiện để có at-least-once.
Ném lỗi thay vì bỏ batch: job khởi động lại và ghi lại từ checkpoint trước. Có thể bị ghi trùng, nhưng không mất, và trùng thì chặng sau xử lý được.
Flush theo thời gian cần một bộ hẹn giờ thật, cùng khoá đồng bộ giữa luồng hẹn giờ và luồng xử lý.
Tốt nhất là không tự viết: connector JDBC chính thức của Flink (JdbcSink với JdbcExecutionOptions) và nền tảng AsyncSinkBase cho sink mới đã xử lý sẵn lô, hẹn giờ, retry và flush khi checkpoint.

Chặng 4: ReplacingMergeTree khử trùng, nhưng muộn

Sink at-least-once nghĩa là ClickHouse sẽ nhận trùng. Bảng dùng ReplacingMergeTree(checkpoint_version), với checkpoint_version là thời điểm kết thúc window, nên cùng một window tính lại cho ra đúng cùng một dòng và cùng version.

Điều cần biết: ReplacingMergeTree khử trùng khi merge các part ở nền, không phải lúc insert. Giữa hai thời điểm đó, SELECT sẽ thấy cả hai bản. Truy vấn đọc phải dùng FINAL, hoặc tự gom theo khoá và lấy version lớn nhất. Phần này dự án làm đúng: các truy vấn đọc bảng deployment và incident, hai bảng ReplacingMergeTree, đều có FINAL.

Chặng 5: materialized view đếm trùng mà merge không sửa được

Đây là chỗ khó thấy nhất. Bảng sự kiện đã chuẩn hoá là ReplacingMergeTree, và một materialized view đổ từ bảng đó vào bảng tổng hợp theo giờ:

CREATE MATERIALIZED VIEW mv_employee_metrics_hourly TO employee_metrics_hourly AS
SELECT org_id, employee_id, toStartOfHour(event_time) AS window_end,
       sumState(toUInt32(event_type = 'pr.merged')) AS pr_merged,
       ...
FROM normalized_events

Materialized view trong ClickHouse là trigger chạy trên mỗi khối được insert, không phải truy vấn trên dữ liệu đã merge. Một sự kiện được insert hai lần thì view thấy hai lần, và bảng tổng hợp cộng hai lần. Khi ReplacingMergeTree gộp hai bản trong bảng gốc, bảng tổng hợp không hề biết. Lớp "idempotent" ở bảng gốc không truyền xuống bảng tổng hợp.

View chạy trên từng khối được insert, nên việc merge ở bảng gốc không bao giờ tới được bảng tổng hợp.

Có vài hướng xử lý, chọn theo nhu cầu:

Tổng hợp trong Flink, ClickHouse chỉ lưu kết quả. Flink đã có window; ghi thẳng kết quả từng window vào một bảng ReplacingMergeTree có khoá là window. Ghi lại bao nhiêu lần cũng ra cùng một dòng.
Khử trùng ngay lúc insert: ClickHouse bỏ được các khối insert giống hệt nhau (mặc định cho bảng replicated, còn bảng thường phải đặt non_replicated_deduplication_window lớn hơn 0), và áp cho cả view phụ thuộc qua deduplicate_blocks_in_dependent_materialized_views (chỉ bật mặc định từ bản 26.2, nên với bản 24.3 mà pipeline này dùng thì phải tự bật). Lần thử lại cần gửi đúng khối cũ, hoặc kèm cùng một insert_deduplication_token, ví dụ dựng từ checkpoint id và subtask.
Tính lại theo lịch từ bảng gốc với FINAL (refreshable materialized view), chấp nhận độ trễ.

Bonus: số 0 không phải lúc nào cũng trung hoà

Cũng trong view đó có vài cột chưa có dữ liệu, được điền giá trị giữ chỗ:

sumState(toUInt32(0))   AS review_comments,       -- vô hại
avgState(toFloat64(0))  AS pr_review_latency_s,   -- không vô hại

Thêm 0 vào một tổng thì tổng không đổi. Thêm 0 vào một trung bình thì mỗi sự kiện góp một mẫu bằng 0, kéo trung bình xuống. Giá trị thật do Flink ghi vào cùng bảng sẽ bị pha loãng bởi hàng nghìn số 0. Đó cũng là lý do AggregatingMergeTree lưu trạng thái trung bình (tổng và số lượng) thay vì lưu chính số trung bình, vì không thể cộng các trung bình với nhau.

Mỗi hàm tổng hợp có phần tử trung hoà riêng: với sum là 0, với avg là "không có mẫu nào". Muốn giữ chỗ cho cột trung bình, hãy tạo một trạng thái rỗng, ví dụ avgStateIf(toFloat64(0), 0), hoặc bỏ cột đó khỏi view cho tới khi có dữ liệu thật.

Tóm tắt

Chặng	Rủi ro	Cách làm đúng
Gateway	Giữ khoá trước khi gửi xong	Chờ Kafka xác nhận rồi mới trả 2xx; chỉ đánh dấu xong khi đã xong
Flink	(làm đúng)	Khử trùng bằng keyed state có TTL, nằm trong checkpoint
Sink	Bộ đệm không gắn với checkpoint	Flush trong `snapshotState`, ném lỗi thay vì bỏ, hoặc dùng sink có sẵn
ClickHouse	Khử trùng khi merge	Đọc bằng `FINAL`
Materialized view	Đếm trùng trước khi merge	Tổng hợp ở Flink, hoặc khử trùng khối insert

Không chặng nào trong số này dùng công nghệ "sai". Lỗi nằm ở chỗ ghép nối, nơi mỗi thành phần giữ đúng lời hứa của riêng nó nhưng không ai giữ lời hứa cho cả chuỗi. Khi thiết kế, mình thấy hữu ích nhất là đi từng mũi tên trong sơ đồ và hỏi: nếu bước này chạy lại, hoặc chết giữa chừng, thì sao?

Một pipeline tương tự có thể còn chỗ hở mà mình chưa thấy. Nếu bạn từng gặp, mình rất muốn được trao đổi. Email của mình ở trang chủ.