Ở Hasaki, khi kho không đủ hàng để đáp ứng một đơn hàng, hệ thống tạo yêu cầu điều chuyển nội bộ để lấy hàng từ địa điểm khác về. Luồng hiện tại publish một Kafka message cho mỗi đơn hàng, và mỗi message lại kích hoạt một yêu cầu điều chuyển riêng.

Vào giờ cao điểm, một kho có thể tạo ra hàng trăm message chỉ trong vài phút. Mỗi message tạo một yêu cầu điều chuyển nhỏ. Đội kho phải xử lý hàng chục yêu cầu lắt nhắt cho cùng một điểm đến, trong khi chỉ cần gom thành một batch là đủ.

Trước:  Đơn A001 → phiếu điều chuyển #1 cho WH-42    (100 phiếu trong 5 phút)
        Đơn A002 → phiếu điều chuyển #2 cho WH-42
        Đơn A003 → phiếu điều chuyển #3 cho WH-42
        ...

Sau:    WH-42 (1 batch đã gộp)                       → hệ thống phía sau

Cách tụi mình xử lý là đặt một debounced aggregator giữa topic chứa event theo từng đơn hàng và các hệ thống phía sau. Aggregator lưu tạm dữ liệu theo kho, rồi publish batch đã gộp khi thỏa mãn các điều kiện về thời gian.

Vấn đề và kiến trúc⌗

Vấn đề cần xử lý⌗

Mỗi event của một đơn hàng lập tức tạo việc cho đội kho. Khi message đến dồn dập, đội kho phải xử lý nhiều yêu cầu điều chuyển nhỏ cho cùng một điểm đến. Các hệ thống phía sau cũng tốn nhiều công xử lý hơn và phải chịu những đợt tải tăng vọt thất thường.

Tụi mình muốn giữ nguyên tính đúng đắn, đồng thời gom event thành batch theo từng kho để các hệ thống phía sau nhận lưu lượng dễ dự đoán hơn.

Luồng tổng quan⌗

Aggregator là một service riêng, consume topic hiện có nên các producer phía trước không cần thay đổi.

Kiến trúc debounced aggregator với hai giai đoạn ingest và flush

Service có hai giai đoạn:

  1. Trong giai đoạn ingest, service nhận Kafka message rồi dùng UPSERT để ghi vào bảng buffer MySQL với trạng thái PENDING.
  2. Trong giai đoạn flush, poller claim các dòng đã sẵn sàng, gộp dữ liệu, gửi một message cho cả batch rồi đánh dấu các dòng là SENT.

Kafka vẫn là nguồn dữ liệu gốc để đối chiếu (source of truth). Bảng MySQL chỉ lưu dữ liệu dẫn xuất làm buffer, giúp kiểm soát việc tạo batch.

Giai đoạn ingest⌗

Tụi mình giữ giai đoạn ingest đơn giản: mỗi bộ (partition_key, location_id, order_id) tương ứng một dòng. Nếu cùng đơn hàng xuất hiện lại khi dòng đó vẫn đang chờ xử lý, service cập nhật payload và timestamp.

INSERT INTO event_buffer_table
  (partition_key, location_id, order_id, payload, status)
VALUES (?, ?, ?, ?, 'PENDING')
ON DUPLICATE KEY UPDATE
  payload = IF(status = 'PENDING', VALUES(payload), payload),
  updated_at = IF(status = 'PENDING', NOW(), updated_at);

Service commit offset thủ công (enable.auto.commit: false), chỉ sau khi ghi database thành công. Nếu service crash trước khi commit offset, Kafka sẽ phát lại message và service có thể xử lý lại an toàn.

Giai đoạn flush⌗

Poller, chẳng hạn chạy mỗi 30 giây, sẽ flush nhóm dữ liệu của một kho khi thỏa mãn một trong hai điều kiện:

  • Debounce idle: Không có cập nhật mới trong 5 phút.
  • Hard window: Dòng đang chờ xử lý lâu nhất đã tồn tại 30 phút.
Cơ chế flush lai giữa debounce idle và hard window

Flush gồm hai bước:

  1. Poller claim các dòng và gán batch_id mới, với số dòng tối đa là max_batch_size.
  2. Service gửi các batch đã claim, rồi đánh dấu các dòng là SENT.
-- Bước 1. Giai đoạn claim: tìm các bucket kho có thể flush (chỉ PENDING)
SELECT location_id
FROM event_buffer_table
WHERE status = 'PENDING'
GROUP BY location_id
HAVING MAX(updated_at) < NOW() - INTERVAL 5 MINUTE
    OR MIN(created_at) < NOW() - INTERVAL 30 MINUTE;

-- Bước 2. Giai đoạn claim: gán batch_id bất biến cho một bucket
SET @batch_id := UUID();
UPDATE event_buffer_table
SET status = 'CLAIMED', batch_id = @batch_id, claimed_at = NOW()
WHERE status = 'PENDING' AND location_id = ?
ORDER BY created_at ASC
LIMIT 500;

-- Bước 3. Giai đoạn flush (worker/tick riêng): đọc toàn bộ row đã claim
SELECT batch_id, payload
FROM event_buffer_table
WHERE status = 'CLAIMED'
ORDER BY claimed_at ASC, batch_id ASC;

-- Bước 4. Xử lý các batch đã claim trong code ứng dụng, rồi đánh dấu các row là SENT.

Các kiểu lỗi và cách xử lý⌗

Các kiểu lỗi ở giai đoạn ingest⌗

Nếu commit Kafka offset trước khi dữ liệu được ghi bền vững vào database thì sao?⌗

Nếu service crash sau khi commit offset nhưng trước khi database commit, message sẽ bị bỏ sót trong quá trình xử lý. Để tránh lỗi này, tụi mình cho service commit offset thủ công, chỉ sau khi ghi database thành công.

Thứ tự commit offset sai và đúng so với thời điểm dữ liệu được lưu bền vững trong database

Nếu service crash trước khi commit offset, Kafka phát lại message. Nếu service crash sau đó, dữ liệu đã được lưu bền vững trong database. Đổi lại, độ trễ xử lý từ đầu đến cuối phụ thuộc vào độ trễ ghi database.

Các kiểu lỗi ở giai đoạn flush⌗

Nếu message dồn quá nhiều trong 30 phút khiến process hết bộ nhớ thì sao?⌗

Nếu chỉ chờ đủ hard window 30 phút, dữ liệu lưu tạm của một kho có thể tăng quá lớn. Khi process chết, việc phục hồi có thể chậm và tạo ra những đợt tải tăng vọt.

Tụi mình dùng debounce idle để flush sớm khi có khoảng lặng 5 phút, đồng thời giới hạn số dòng mỗi lần claim bằng max_batch_size.

Lưu lượng dồn dập được chia thành các lần flush nhỏ hơn nhờ debounce và claim có giới hạn

Nhờ vậy, công việc được chia thành nhiều batch nhỏ, độ trễ được giữ trong giới hạn và áp lực bộ nhớ giảm. Đổi lại, flush thường xuyên hơn có thể làm giảm nhẹ lượng dữ liệu gom được trong mỗi batch.

Nếu message của một kho đến liên tục khiến debounce không bao giờ kích hoạt thì sao?⌗

Dù message chỉ đến nhỏ giọt, nếu đến liên tục thì updated_at cứ được cập nhật và điều kiện idle không bao giờ đúng. Các dòng có thể chờ vô thời hạn.

Tụi mình giữ hard window 30 phút tính từ dòng đang chờ xử lý lâu nhất.

Luồng message liên tục, hard window buộc phải flush dù không có khoảng lặng nào

Nhờ vậy, dữ liệu của từng kho vẫn được xử lý tiếp dù không có khoảng lặng. Debounce giúp tối ưu việc gom batch khi có khoảng lặng, còn hard window đảm bảo việc xử lý vẫn tiếp tục khi message đến liên tục. Đặc tính này được gọi là liveness. Đổi lại, một số lần flush diễn ra trước khi batch đạt kích thước lý tưởng.

Nếu gửi thành công nhưng service crash trước khi đánh dấu đã gửi, rồi chạy lại thì sao?⌗

Nếu batch không có định danh ổn định, mỗi lần retry có thể gom đơn hàng khác nhau, chẳng hạn lần đầu A với B, lần sau A với C. Điều này gây khó khăn cho việc giữ dữ liệu nhất quán ở các hệ thống phía sau.

Tụi mình gán batch_id lúc claim và giữ nguyên cho đến khi trạng thái là SENT. Khi retry, service gửi lại đúng các dòng đã claim với cùng batch_id.

Crash sau khi publish nhưng trước khi đánh dấu đã gửi, và lần retry dùng lại cùng batch_id

Consumer phía sau dùng batch_id làm idempotency key để loại bỏ message trùng. Cách này giữ được at-least-once delivery mà tránh thực hiện nghiệp vụ lặp lại, nhưng đòi hỏi consumer phía sau phải loại bỏ message trùng theo batch_id.

Nếu query phục vụ flush chạy chậm khi tải cao thì sao?⌗

Flush lượng dữ liệu lớn mà không có giới hạn làm tăng thời gian giữ lock, độ trễ query và rủi ro ở những lần xử lý chậm nhất.

Tụi mình tách claim và flush thành hai bước, giới hạn số dòng mỗi lần claim bằng max_batch_size, và retry từ trạng thái CLAIMED. Nếu Kafka tạm thời không khả dụng sau khi claim, các dòng vẫn giữ trạng thái CLAIMED để được xử lý lại ở tick tiếp theo.

Mỗi tick chỉ xử lý lượng công việc có giới hạn, còn các batch đã claim nhưng chưa gửi được sẽ được retry. Đổi lại, có thể cần nhiều tick mới xử lý hết dữ liệu của một kho.

Nếu nhiều pod cùng chạy job flush thì sao?⌗

Hai pod có thể claim và flush đồng thời, khiến thứ tự claim, mức độ tranh chấp và thời điểm retry không được xác định rõ. Isolation level cao hơn như repeatable read có thể giảm các bất thường, nhưng vẫn chưa thể hiện rõ yêu cầu chỉ có một bên chạy job trên toàn hệ thống.

Trước mỗi tick flush, mọi pod đều thử lấy Redis lock. Chỉ pod giữ lock mới chạy job; các pod còn lại bỏ qua tick đó. Lock có TTL 60 giây, còn job có timeout 54 giây, tức 90% TTL. Khoảng chênh 6 giây dự phòng cho sai lệch đồng hồ, độ trễ lập lịch và độ trễ mạng.

Cơ chế này đảm bảo tại một thời điểm chỉ có một flush worker hoạt động giữa các pod, đồng thời tự phục hồi nếu pod crash. Timeout phải ngắn hơn TTL. Nếu cả hai đều là 60 giây, chỉ cần lệch 10 ms là lock có thể hết hạn trước khi job đầu tiên dừng. Pod khác khi đó có thể lấy lock và bắt đầu flush lần thứ hai.

Mốc 90% là một heuristic, không phải đảm bảo tuyệt đối. Process bị tạm dừng lâu hoặc độ trễ mạng lớn vẫn có thể vượt khoảng dự phòng ấy. Tần suất flush cũng phụ thuộc vào tính sẵn sàng của Redis.

Nếu bảng buffer cứ tăng mãi thì sao?⌗

Dùng DELETE hàng loạt để dọn dữ liệu cũ có thể gây phân mảnh InnoDB nghiêm trọng và làm giảm thông lượng.

Tụi mình dùng range partitioning theo tháng trong MySQL và drop partition theo lịch. Trước khi drop, phải kiểm tra partition không còn dòng nào chưa gửi.

Chiến lược dọn dẹp bằng cách drop partition thay vì delete hàng loạt

Drop partition là thao tác trên metadata, giúp tránh phải xóa lượng dữ liệu lớn lần lượt từng dòng. Đổi lại, quản lý partition đòi hỏi quy trình vận hành chặt chẽ và các bước kiểm tra an toàn.

Quy ước với hệ thống phía sau⌗

Hệ thống phía sau consume một message đã gộp cho mỗi lần flush của một kho:

  • Key: location_id
  • Định danh: batch_id
  • Payload: order_ids, items đã gộp và flushed_at

Hệ thống phía sau phải loại bỏ message trùng theo batch_id để giữ việc xử lý idempotent.