Phần 2: Kiến Trúc Hướng Sự Kiện — Quản Trị Kafka Siêu Quy Mô, Transactional Outbox & Xử Lý Bất Biến

📖 Bản tiếng Anh (English Edition)

Bài trước: Phần 1 — Nền Tảng Microservices & Tự Động Hóa GitOps | Danh mục Series | Bài tiếp: Phần 3 — Tầng Dữ Liệu: Chuyển Đổi Từ Aurora Sang TiDB

Answer-first: PayPay triệt tiêu rủi ro mất dữ liệu và phân tách thanh toán ở đỉnh 1.250 TPS nhờ Transactional Outbox cùng Debezium CDC và Apache Kafka. Nhóm consumer dùng CooperativeStickyAssignor loại bỏ điểm dừng rebalance, kết hợp khóa phân tán Redis hai giai đoạn bảo đảm xử lý chính xác một lần trước khi cập nhật sổ cái database.

Điều kiện tiên quyết: Nắm vững cấu trúc phân vùng nhật ký Apache Kafka, giao thức consumer group, kỹ thuật Change Data Capture (CDC) với Debezium và cơ chế khóa phân tán.

1. Rủi Ro Ghi Kép (Dual-Write Hazard) Trong Hệ Thống Tài Chính Tải Cao

Trong kỹ nghệ phát triển nền tảng thanh toán, một trong những cạm bẫy kiến trúc nguy hiểm nhất là Rủi Ro Ghi Kép (Dual-Write Hazard). Khi một yêu cầu thanh toán từ người dùng gửi tới, microservice thanh toán bắt buộc phải thực hiện song song hai tác vụ: thay đổi trạng thái trong cơ sở dữ liệu quan hệ cục bộ (trừ số dư ví, chốt dòng sổ cái kế toán kép) và bắn thông báo tới các dịch vụ downstream (trả thưởng hoàn tiền cashback, gửi thông báo đẩy biến động số dư, chấm điểm rủi ro gian lận).

Nếu lập trình viên cố gắng thực hiện đồng thời hai thao tác này trong mã nguồn ứng dụng, hệ thống sẽ rơi vào tình thế bế tắc về tính nhất quán phân tán:

Kịch Bản A (Commit Cơ Sở Dữ Liệu Trước):
1. DB.Begin()
2. UPDATE wallet SET balance = balance - 1000 WHERE id = 42;
3. INSERT INTO payments (id, amount) VALUES ('pay-101', 1000);
4. DB.Commit() -> THÀNH CÔNG!
5. kafkaProducer.Send(PaymentCompletedEvent) -> LỖI MẠNG TIMEOUT HOẶC CRASH!
Hậu quả: Tiền của người dùng đã bị trừ, nhưng thông báo và tiền hoàn cashback bị biến mất vĩnh viễn!

Kịch Bản B (Bắn Sự Kiện Sang Kafka Trước):
1. kafkaProducer.Send(PaymentCompletedEvent) -> THÀNH CÔNG!
2. DB.Begin()
3. UPDATE wallet SET balance = balance - 1000 WHERE id = 42;
4. DB.Commit() -> DEADLOCK HOẶC TRÙNG KHÓA DUY NHẤT! -> ROLLBACK!
Hậu quả: Các consumer hạ nguồn đã trả thưởng và nhả hàng cho một đơn thanh toán chưa từng thành công trong sổ cái!

Trong môi trường phân tán quy mô lớn, việc sử dụng cơ chế Two-Phase Commit (2PC) truyền thống bị bác bỏ hoàn toàn do độ trễ quá cao, giữ lock tài nguyên lâu và phụ thuộc vào điểm nghẽn điều phối viên (coordinator). Thay vào đó, việc bảo đảm tính nhất quán giữa cơ sở dữ liệu và hàng đợi message đòi hỏi một mẫu hình chuyển trạng thái nguyên tử từ gốc.

2. Mô Hình Transactional Outbox Kết Hợp Debezium CDC

PayPay giải quyết triệt để bài toán ghi kép bằng cách triển khai mô hình Transactional Outbox kết hợp công nghệ Change Data Capture (CDC) cấp log thông qua Debezium và Apache Kafka:

flowchart TD
    subgraph PaymentService["Dịch Vụ Thanh Toán Cốt Lõi (Go 1.25 Engine)"]
        CLIENT["Ứng Dụng Di Động Quét Mã Thanh Toán"]
        BIZ_TX["Giao Dịch ACID Nguyên Tử Cục Bộ:<br/>1. Trừ Số Dư Ví Người Dùng<br/>2. Ghi Dòng Sổ Cái Kế Toán Kép<br/>3. Chèn Bản Ghi Sự Kiện Vào Bảng Outbox"]
    end

    subgraph DatabaseTier["Tầng Cơ Sở Dữ Liệu Phân Tán (TiDB / MySQL)"]
        LEDGER_TBL["Bảng: wallet_ledger"]
        OUTBOX_TBL["Bảng: outbox_events"]
        WAL["Nhật Ký Giao Dịch Ghi Trước (WAL / MySQL Binlog)"]
    end

    subgraph IngestionPipeline["Cụm Pipeline Thu Thập Dữ Liệu CDC"]
        DEBEZIUM["Cụm Debezium Kafka Connect<br/>(Cơ Chế Dự Phòng Failover)"]
    end

    subgraph EventStream["Cụm Hàng Đợi Apache Kafka (AWS Tokyo)"]
        TOPIC_PAY["Topic: payment-events<br/>(64 Partitions, Khóa Phân Vùng: user_id)"]
    end

    subgraph DownstreamConsumers["Các Consumer Nghiệp Vụ Độc Lập"]
        CONS_NOTIF["Dịch Vụ Thông Báo Đẩy / SMS"]
        CONS_REWARD["Cỗ Máy Tính Thưởng Chiến Dịch Cashback"]
        CONS_RISK["Dịch Vụ Kiểm Toán & Chống Rửa Tiền AML"]
    end

    CLIENT --> BIZ_TX
    BIZ_TX --> LEDGER_TBL
    BIZ_TX --> OUTBOX_TBL
    OUTBOX_TBL --> WAL
    WAL --> DEBEZIUM
    DEBEZIUM --> TOPIC_PAY

    TOPIC_PAY --> CONS_NOTIF
    TOPIC_PAY --> CONS_REWARD
    TOPIC_PAY --> CONS_RISK

Nguyên Lý Vận Hành Và Ràng Buộc Bất Biến

1. Giao Dịch Cục Bộ Nguyên Tử: Khi người dùng xác thực thanh toán, dịch vụ mở một transaction ACID tiêu chuẩn trên cơ sở dữ liệu. Nó thực hiện đồng thời việc trừ số dư, ghi nhận dòng sổ cái và chèn một bản ghi sự kiện vào bảng outbox_events trong cùng một giao dịch duy nhất. Hoặc toàn bộ cả 3 thao tác cùng được ghi xuống đĩa, hoặc không có thao tác nào được thực thi.
2. Thu Thập Bất Đồng Bộ Cấp Nhật Ký (Log-Based CDC): Thay vì sử dụng worker chạy các câu lệnh SELECT * FROM outbox_events WHERE processed = false quét định kỳ (gây table lock và nghẽn CPU cơ sở dữ liệu), Debezium kết nối trực tiếp với vai trò replica đọc stream binary log (MySQL binlog hoặc TiKV CDC stream). Debezium giải mã các thay đổi vừa commit trực tiếp từ log mà không tiêu tốn tài nguyên truy vấn của database.
3. Định Tuyến Khóa Phân Vùng Bảo Đảm Thứ Tự: Debezium đóng gói payload sự kiện và bắn vào topic payment-events của Kafka với Partition Key là user_id hoặc wallet_id. Do Kafka đảm bảo tuyệt đối tính tuần tự FIFO trên từng phân vùng, mọi giao dịch thuộc về cùng một tài khoản người dùng sẽ luôn được xử lý theo đúng trình tự thời gian phát sinh.

Bảng so sánh chi tiết giữa Log-Based Outbox và các phương án tiếp cận khác:

3. Cơ Chế Rebalance Của Consumer Group: Eager Đối Đầu Cooperative Sticky

Trong các hệ thống streaming quy mô lớn, việc tự động co giãn (autoscaling) hạm đội pod consumer là bắt buộc để hấp thụ các đợt lưu lượng bùng nổ trong chiến dịch khuyến mãi. Tuy nhiên, việc cấu hình sai giải thuật rebalance trên Kafka có thể gây ra Cơn Bão Rebalance (Rebalance Storm), khiến toàn bộ cụm consumer dừng hoạt động và lượng message tồn đọng (lag) tăng phi mã.

flowchart TD
    subgraph EagerRebalance["Cơ Chế Eager Rebalance Cũ (Dừng Toàn Hệ Thống - Stop-the-World)"]
        E_T1["T0: Nhóm Consumer có 3 pod đang chia nhau 64 partition"]
        E_T2["T1: Autoscale kích hoạt: Pod thứ 4 tham gia nhóm"]
        E_T3["T2: STOP-THE-WORLD! Toàn bộ 3 pod buông TẤT CẢ các partition"]
        E_T4["T3: Group Coordinator tính toán lại toàn bộ sơ đồ phân chia"]
        E_T5["T4: Cả 4 pod kết nối lại và nạp lại trạng thái (đình trệ 2-10 giây)"]
        E_T1 --> E_T2 --> E_T3 --> E_T4 --> E_T5
    end

    subgraph CooperativeSticky["Cơ Chế Cooperative Sticky Rebalance (Tăng Dần & Không Gián Đoạn)"]
        C_T1["T0: Nhóm Consumer có 3 pod đang chia nhau 64 partition"]
        C_T2["T1: Autoscale kích hoạt: Pod thứ 4 tham gia nhóm"]
        C_T3["T2: Chỉ các partition cần điều chuyển (ví dụ 16 partition) bị thu hồi"]
        C_T4["T3: 48 partition không bị ảnh hưởng VẪN TIẾP TỤC XỬ LÝ bình thường!"]
        C_T5["T4: Pod thứ 4 tiếp nhận 16 partition được bàn giao một cách mượt mà"]
        C_T1 --> C_T2 --> C_T3 --> C_T4 --> C_T5
    end

Triệt Tiêu Độ Trễ Đình Trệ Toàn Hệ Thống

Với giao thức rebalance truyền thống (sử dụng RangeAssignor hoặc RoundRobinAssignor), mỗi khi có sự biến động thành viên — chẳng hạn một pod khởi động lại, thất bại liveness probe hoặc cụm tự động scale up — toàn bộ các consumer trong nhóm buộc phải hủy bỏ quyền xử lý trên tất cả các partition đang giữ. Quá trình tiêu thụ dữ liệu bị đóng băng hoàn toàn cho đến khi Kafka Group Coordinator hoàn tất phân bổ và tất cả thành viên đồng thuận nhận partition mới. Ở lưu lượng tải đỉnh (trên 1.250 TPS), việc đình trệ 5 giây sẽ tích tụ hơn 6.000 bản tin chưa xử lý, kích hoạt cảnh báo trễ và dẫn tới sự cố sập dây chuyền.

PayPay loại bỏ hoàn toàn rủi ro này bằng cách bắt buộc áp dụng CooperativeStickyAssignor:

• Tái Cân Bằng Tăng Dần Hai Giai Đoạn (Incremental Rebalancing): Thay vì thu hồi hàng loạt, các consumer vẫn tiếp tục xử lý các partition không thay đổi quyền sở hữu trong suốt quá trình rebalance. Chỉ những partition được chọn để điều chuyển sang pod mới mới tạm thời bị thu hồi và bàn giao.
• Tính Bám Dính Của Partition (Sticky Affinity): Các partition được ưu tiên giữ nguyên gắn kết với pod cũ. Điều này bảo vệ bộ nhớ đệm in-memory cục bộ, giảm tải dọn rác (GC) và loại bỏ hoàn toàn tình trạng đình trệ xử lý trong các đợt co giãn pod.

4. Xử Lý Bất Biến (Idempotency) & Ngữ Nghĩa Chính Xác Một Lần

Do sự cố mạng giữa các consumer và Kafka broker có thể khiến thao tác commit offset thất bại, Kafka chỉ cung cấp cam kết phân phối ít nhất một lần (at-least-once delivery). Một pod consumer có thể bị crash ngay sau khi cập nhật dữ liệu vào sổ cái nhưng chưa kịp báo nhận offset lên Kafka, khiến Kafka chuyển giao lại chính bản tin đó cho một pod khác xử lý lại. Do đó, các dịch vụ tiêu thụ bắt buộc phải triển khai cơ chế xử lý bất biến (idempotency) để ngăn chặn việc trừ tiền hoặc cộng thưởng hai lần.

sequenceDiagram
    autonumber
    participant Kafka as Phân Vùng Kafka (payment-events)
    participant Worker as Worker Xử Lý Thanh Toán (Go 1.25)
    participant Redis as Cụm Redis Sentinel (Bộ Nhớ Đệm Idempotency)
    participant DB as Cụm Sổ Cái Tài Chính TiDB
    participant DLQ as Hàng Đợi Chết (payment-events-dlq)

    Kafka->>Worker: Chuyển giao Message (event_id="ev-7701", amount=5000)
    Worker->>Redis: Kiểm tra processed:event:ev-7701
    alt Bản Tin Đã Được Xử Lý Trước Đó
        Redis-->>Worker: Tồn Tại (Trạng thái: "COMPLETED")
        Worker->>Kafka: Commit Offset (Xác nhận & Bỏ qua an toàn)
    else Bản Tin Mới Nhận Lần Đầu
        Redis-->>Worker: Khóa Không Tồn Tại
        Worker->>Redis: SET lock:idempotency:ev-7701 "IN_FLIGHT" NX EX 120
        alt Nhận Được Khóa Phân Tán
            Redis-->>Worker: OK (Cấp Quyền Xử Lý)
            Worker->>DB: Mở Giao Dịch ACID (Ghi Sổ Cái + Cập Nhật Số Dư)
            alt Giao Dịch Database Thành Công
                DB-->>Worker: Commit Thành Công
                Worker->>Redis: SET processed:event:ev-7701 "COMPLETED" EX 86400
                Worker->>Redis: DEL lock:idempotency:ev-7701
                Worker->>Kafka: Commit Offset Message
            else Lỗi Tạm Thời Cơ Sở Dữ Liệu / Deadlock
                DB-->>Worker: Lỗi: Deadlock / Timeout
                Worker->>Redis: DEL lock:idempotency:ev-7701
                Note over Worker: Giải phóng khóa để pod khác có thể thử lại ngay
            end
        else Đụng Độ Khóa (Có Worker Khác Đang Xử Lý Song Song)
            Redis-->>Worker: Nil (Khóa Đang Bị Giữ)
            Note over Worker: Tạm hoãn chờ worker trước hoàn thành hoặc hết TTL
        end
    else Lỗi Dữ Liệu Độc Hại / Payload Hỏng
        Worker->>DLQ: Bắn sang DLQ kèm Stack Trace Chi Tiết
        Worker->>Kafka: Commit Offset (Giải phóng phân vùng)
    end

Mã Nguồn Go 1.25+ Consumer Group Chuẩn Môi Trường Sản Xuất

Đoạn mã Golang 1.25+ dưới đây triển khai giao diện sarama.ConsumerGroupHandler với chiến lược CooperativeStickyAssignor, quản lý log có cấu trúc slog, cơ chế khóa phân tán Redis 2 giai đoạn, giao dịch database thực thụ có kiểm tra phiên bản và định tuyến sang Dead Letter Queue khi gặp lỗi:

// Package consumer cung cấp logic xử lý sự kiện thanh toán bất biến chuẩn production cho PayPay.
package consumer

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"github.com/IBM/sarama"
	"github.com/redis/go-redis/v9"
)

type PaymentEvent struct {
	EventID   string    `json:"event_id"`
	PaymentID string    `json:"payment_id"`
	UserID    int64     `json:"user_id"`
	Amount    int64     `json:"amount"` // Đơn vị Yên Nhật (số nguyên, tránh sai số số thực)
	Timestamp time.Time `json:"timestamp"`
}

type ConsumerGroupHandler struct {
	logger      *slog.Logger
	redisClient *redis.Client
	db          *sql.DB
	dlqProducer sarama.SyncProducer
	dlqTopic    string
}

func NewConsumerGroupHandler(
	logger *slog.Logger,
	redisClient *redis.Client,
	db *sql.DB,
	dlqProducer sarama.SyncProducer,
	dlqTopic string,
) *ConsumerGroupHandler {
	return &ConsumerGroupHandler{
		logger:      logger,
		redisClient: redisClient,
		db:          db,
		dlqProducer: dlqProducer,
		dlqTopic:    dlqTopic,
	}
}

func (h *ConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error {
	h.logger.Info("thiết lập thành công phiên consumer group sarama")
	return nil
}

func (h *ConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error {
	h.logger.Info("hoàn tất dọn dẹp tài nguyên phiên consumer group")
	return nil
}

func (h *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for {
		select {
		case msg, ok := <-claim.Messages():
			if !ok {
				return nil
			}

			ctx, cancel := context.WithTimeout(session.Context(), 10*time.Second)
			err := h.processMessageWithIdempotency(ctx, msg)
			cancel()

			if err != nil {
				h.logger.Error("phát sinh lỗi không thể tự phục hồi, chuyển hướng sang DLQ",
					slog.String("topic", msg.Topic),
					slog.Int("partition", int(msg.Partition)),
					slog.Int64("offset", msg.Offset),
					slog.String("error", err.Error()),
				)
				h.routeToDLQ(msg, err)
			}

			session.MarkMessage(msg, "")

		case <-session.Context().Done():
			return nil
		}
	}
}

func (h *ConsumerGroupHandler) processMessageWithIdempotency(ctx context.Context, msg *sarama.ConsumerMessage) error {
	var event PaymentEvent
	if err := json.Unmarshal(msg.Value, &event); err != nil {
		return fmt.Errorf("payload JSON hỏng: %w", err)
	}

	if event.EventID == "" || event.UserID == 0 {
		return errors.New("schema sự kiện không hợp lệ: thiếu event_id hoặc user_id")
	}

	processedKey := fmt.Sprintf("processed:event:%s", event.EventID)
	lockKey := fmt.Sprintf("lock:idempotency:%s", event.EventID)

	// Bước 1: Kiểm tra xem sự kiện đã hoàn thành trước đó chưa (cửa sổ lưu giữ 24 giờ)
	exists, err := h.redisClient.Exists(ctx, processedKey).Result()
	if err != nil {
		return fmt.Errorf("lỗi kiểm tra redis: %w", err)
	}
	if exists > 0 {
		h.logger.DebugContext(ctx, "phát hiện bản tin trùng lặp, bỏ qua thực thi an toàn",
			slog.String("event_id", event.EventID))
		return nil
	}

	// Bước 2: Chiếm khóa phân tán tạm thời (thời hạn thuê 2 phút)
	acquired, err := h.redisClient.SetNX(ctx, lockKey, "IN_FLIGHT", 2*time.Minute).Result()
	if err != nil {
		return fmt.Errorf("lỗi chiếm khóa phân tán redis: %w", err)
	}
	if !acquired {
		return fmt.Errorf("worker khác đang xử lý sự kiện %s, thử lại sau", event.EventID)
	}
	defer h.redisClient.Del(ctx, lockKey)

	// Bước 3: Thực thi cập nhật số dư trong giao dịch database tường minh
	tx, err := h.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return fmt.Errorf("không thể mở giao dịch database: %w", err)
	}
	defer tx.Rollback()

	// Cập nhật số dư người dùng kèm kiểm tra phiên bản lạc quan (optimistic locking)
	query := `
		UPDATE wallet_balances 
		SET balance = balance + ?, updated_at = NOW(), version = version + 1 
		WHERE user_id = ?`
	res, err := tx.ExecContext(ctx, query, event.Amount, event.UserID)
	if err != nil {
		return fmt.Errorf("lỗi cập nhật số dư ví: %w", err)
	}

	rowsAffected, err := res.RowsAffected()
	if err != nil || rowsAffected == 0 {
		return fmt.Errorf("không tìm thấy bản ghi ví cho user_id=%d", event.UserID)
	}

	// Ghi nhật ký kiểm toán vào sổ nhật ký tài chính kép
	auditQuery := `
		INSERT INTO payment_audit_journal (event_id, payment_id, user_id, amount, status, created_at)
		VALUES (?, ?, ?, ?, 'SETTLED', NOW())`
	if _, err := tx.ExecContext(ctx, auditQuery, event.EventID, event.PaymentID, event.UserID, event.Amount); err != nil {
		return fmt.Errorf("không thể ghi bản ghi kiểm toán thanh toán: %w", err)
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("không thể commit giao dịch sổ cái: %w", err)
	}

	// Bước 4: Đánh dấu trạng thái hoàn tất vĩnh viễn vào Redis cache
	if err := h.redisClient.Set(ctx, processedKey, "COMPLETED", 24*time.Hour).Err(); err != nil {
		h.logger.WarnContext(ctx, "không thể lưu khóa hoàn tất idempotency vào redis",
			slog.String("event_id", event.EventID), slog.String("error", err.Error()))
	}

	return nil
}

func (h *ConsumerGroupHandler) routeToDLQ(msg *sarama.ConsumerMessage, processErr error) {
	dlqMsg := &sarama.ProducerMessage{
		Topic: h.dlqTopic,
		Key:   sarama.ByteEncoder(msg.Key),
		Value: sarama.ByteEncoder(msg.Value),
		Headers: []sarama.RecordHeader{
			{Key: []byte("original-topic"), Value: []byte(msg.Topic)},
			{Key: []byte("error-message"), Value: []byte(processErr.Error())},
			{Key: []byte("failed-at"), Value: []byte(time.Now().UTC().Format(time.RFC3339))},
		},
	}
	if _, _, err := h.dlqProducer.SendMessage(dlqMsg); err != nil {
		h.logger.Error("lỗi nghiêm trọng: không thể gửi bản tin vào hàng đợi chết DLQ",
			slog.String("dlq_topic", h.dlqTopic), slog.String("error", err.Error()))
	}
}

// BuildCooperativeConsumerConfig cấu hình client Sarama cụm production.
func BuildCooperativeConsumerConfig() *sarama.Config {
	config := sarama.NewConfig()
	config.Version = sarama.V3_6_0_0
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{
		sarama.NewBalanceStrategyCooperativeSticky(),
	}
	config.Consumer.Group.Session.Timeout = 20 * time.Second
	config.Consumer.Group.Heartbeat.Interval = 6 * time.Second
	config.Consumer.MaxProcessingTime = 15 * time.Second
	config.Consumer.Return.Errors = true
	return config
}

5. Đánh Đổi Kiến Trúc & Kinh Nghiệm Tối Ưu Môi Trường Sản Xuất

Vận hành đường ống xử lý luồng sự kiện tài chính ở thông lượng cao đòi hỏi các quyết định đánh đổi kỹ thuật nghiêm ngặt:

Câu Hỏi Thường Gặp (FAQ)

📖 Bản tiếng Anh (English Edition)

Bài trước: Phần 1 — Nền Tảng Microservices & Tự Động Hóa GitOps | Danh mục Series | Bài tiếp: Phần 3 — Tầng Dữ Liệu: Chuyển Đổi Từ Aurora Sang TiDB