Kafka Event Streaming
1. Tổng quan
BANA dùng Apache Kafka cho giao tiếp bất đồng bộ, hướng sự kiện giữa các microservice. Sự kiện được publish khi có thay đổi trạng thái nghiệp vụ (ví dụ thanh toán thành công, tạo merchant) và được consume bởi các dịch vụ phía dưới phản ứng với những thay đổi đó.
Tất cả tên topic được định nghĩa tập trung trong @nx/core tại src/queues/kafka/topics.ts.
2. Danh mục Topic
Quy ước tên topic: nx.bana.<evt|cmd|cdc|dlq>.<domain>[.<entity>].<event>. Registry sự kiện ứng dụng (ServiceTopicDefs trong packages/core/src/common/kafka/registry.ts):
| Topic | Loại | Mô tả |
|---|---|---|
nx.bana.evt.payment.succeeded | evt | Thanh toán bán hàng hoàn tất |
nx.bana.evt.inventory.purchase-order.received | evt | Đơn mua được nhận vào kho |
nx.bana.evt.inventory.stock.issued-for-sale | evt | Xuất kho cho bán hàng |
nx.bana.evt.inventory.stock.adjusted | evt | Điều chỉnh tồn kho |
nx.bana.evt.kitchen.ticket-item.status-changed | evt | Trạng thái mục phiếu bếp thay đổi |
nx.bana.evt.inventory.material.transferred | evt | Chuyển nguyên vật liệu giữa các kho |
nx.bana.evt.inventory.material.stock-changed | evt | Tồn nguyên vật liệu thay đổi |
nx.bana.evt.signal.activity.notified | evt | Thông báo activity (→ WebSocket) |
nx.bana.cmd.ledger.generate | cmd | Tạo tài liệu ledger (nội bộ ledger) |
Thay đổi merchant, product và product-variant không phải evt topic - chúng lan truyền dưới dạng CDC Debezium
nx.bana.cdc.<schema>.<Table>(xem CDC). Không có topiccommerce.initialized.
3. Ma trận Producer & Consumer
| Topic | Producer | Consumer |
|---|---|---|
nx.bana.evt.payment.succeeded | sale | finance, inventory, invoice |
nx.bana.evt.inventory.purchase-order.received | inventory | finance |
nx.bana.evt.inventory.stock.issued-for-sale | inventory | finance |
nx.bana.evt.inventory.stock.adjusted | inventory | finance |
nx.bana.evt.kitchen.ticket-item.status-changed | sale | inventory |
nx.bana.evt.inventory.material.transferred | inventory | inventory (chuyển kho) |
nx.bana.evt.inventory.material.stock-changed | inventory | (chưa có consumer) |
nx.bana.evt.signal.activity.notified | (chưa có producer) | signal |
nx.bana.cmd.ledger.generate | ledger | ledger (nội bộ) |
4. Sơ đồ Luồng Message
5. Consumer Group
| Dịch vụ | Client ID | Group ID |
|---|---|---|
| Finance | SVC-00040-FINANCE_CONSUMER | SVC-00040-FINANCE_CONSUMER_GROUP |
| Inventory | SVC-00050-INVENTORY_CONSUMER | SVC-00050-INVENTORY_CONSUMER_GROUP |
| Ledger | SVC-00060-LEDGER_CONSUMER | SVC-00060-LEDGER_GROUP |
| Search (CDC) | cdc-consumer | cdc-consumer-group |
6. Cấu hình Consumer
6.1. Consumer Finance
| Cài đặt | Giá trị |
|---|---|
| Topic | nx.bana.evt.payment.succeeded, …inventory.purchase-order.received, …inventory.stock.issued-for-sale, …inventory.stock.adjusted, CDC public.Merchant |
| Auto-commit | bật |
| Fallback Mode | earliest |
| Deserializer | Key dạng String, Value dạng JSON |
6.2. Consumer Inventory
| Cài đặt | Giá trị |
|---|---|
| Topic | nx.bana.evt.payment.succeeded, …kitchen.ticket-item.status-changed, …inventory.material.transferred, CDC public.Merchant, CDC public.ProductVariant |
| Fallback Mode | earliest |
6.3. Consumer Ledger
| Cài đặt | Giá trị |
|---|---|
| Topic | nx.bana.cmd.ledger.generate |
| Auto-commit | false (commit thủ công sau khi xử lý) |
| Số lượng Consumer | Có thể cấu hình qua APP_ENV_KAFKA_CONSUMER_COUNT |
| Acks (Producer) | ALL |
| Idempotent (Producer) | true |
6.4. Consumer Search CDC
| Cài đặt | Giá trị |
|---|---|
| Topic | 27 topic CDC Debezium |
| Auto-commit | false (quản lý offset thủ công) |
| Max Batch Size | 200 |
| Flush Interval | 2000ms |
| Max Wait Time | 500ms |
| Max Bytes | 5MB |
Xem CDC / Debezium để biết chi tiết.
7. Ví dụ Payload Sự kiện
7.1. Payment Success (nx.bana.evt.payment.succeeded)
json
{
"saleOrderId": "987654321",
"saleOrderNumber": "SO-0001",
"merchantId": "123456789",
"saleChannelId": "555",
"inventoryLocationId": "loc-1",
"items": [{ "id": "i1", "itemType": "PRODUCT_VARIANT", "itemId": "pv-1", "quantity": 2, "mode": "DEDUCT" }]
}7.2. Ledger Generate (nx.bana.cmd.ledger.generate)
json
{
"ledgerId": "777888999",
"ledgerType": "S1a-HKD",
"merchantId": "444555666",
"period": "2026-Q1",
"version": 1,
"isRetry": false,
"enqueueTime": 1711785600000
}8. Thực hành Tốt nhất
8.1. Idempotency
Tất cả consumer đều thực hiện kiểm tra idempotency để xử lý an toàn các message trùng lặp:
| Dịch vụ | Chiến lược |
|---|---|
| Finance | Kiểm tra giao dịch hiện có theo merchantId + referenceType + referenceId |
| Inventory | Kiểm tra vị trí mặc định hiện có theo merchantId |
| Ledger | Kiểm tra tác vụ đang chờ hiện có theo ledgerId |
8.2. Xử lý Lỗi
| Chiến lược | Được dùng bởi |
|---|---|
| Bỏ qua + Log | Finance (log lỗi, tiếp tục xử lý) |
| DLQ (Dead Letter Queue) | Search CDC (nx.bana.dlq.cdc) |
| Retry qua API | Ledger (endpoint retry thủ công) |
| Phục hồi khi khởi động | Ledger (đưa lại các tác vụ bị treo vào hàng đợi) |
8.3. Serialization
| Hướng | Định dạng |
|---|---|
| Producer | JSON serializer (value), string serializer (key) |
| Consumer | JSON deserializer (value), string deserializer (key) |
9. Hạ tầng
9.1. Phát triển (Docker Compose)
Cluster Kafka 3 node:
nx-kafka-1:29092nx-kafka-2:29092nx-kafka-3:29092
9.2. Biến Môi trường
| Biến | Mô tả |
|---|---|
APP_ENV_KAFKA_BROKERS | Địa chỉ broker cách nhau bằng dấu phẩy |
APP_ENV_KAFKA_CLIENT_ID | Định danh client riêng cho dịch vụ |
APP_ENV_KAFKA_GROUP_ID | Định danh consumer group |
APP_ENV_KAFKA_CONSUMER_COUNT | Số consumer song song (Ledger) |
APP_ENV_KAFKA_SASL_MECHANISM | Cơ chế SASL (ví dụ SCRAM-SHA-256) |
APP_ENV_KAFKA_SASL_USERNAME | Username SASL |
APP_ENV_KAFKA_SASL_PASSWORD | Password SASL |
10. Tài liệu Liên quan
| Tài liệu | Mô tả |
|---|---|
| CDC / Debezium | Change Data Capture qua Kafka |
| Finance Service | Consumer sự kiện Finance |
| Inventory Service | Consumer sự kiện Inventory |
| Ledger Service | Pipeline tạo Ledger |
| Search Service | Consumer CDC để đồng bộ Typesense |