Đơn vị biên soạn: TechData.AI
Chủ đề: Data Engineering và Event Streaming
Đối tượng: Data Engineer, Python Developer và Backend Developer
Phiên bản tham chiếu: Apache Kafka 4.x

MỤC LỤC

1. Mục tiêu bài học
2. Kafka Producer làm gì?
3. Chuẩn bị project Python
4. Producer đầu tiên
5. Serialization và cấu trúc Event
6. Message Key và cách chọn Partition
7. Gửi dữ liệu bất đồng bộ
8. Delivery Callback và xử lý lỗi
9. acks, retry và độ tin cậy
10. Idempotent Producer
11. Batch, linger và compression
12. Thiết kế Event chuẩn doanh nghiệp
13. Bảo mật thông tin kết nối
14. Logging, metrics và kiểm thử
15. Project Producer đơn hàng hoàn chỉnh
16. Những lỗi thường gặp
17. Tổng kết
18. Câu hỏi ôn tập
19. Tài liệu tham khảo

1. Mục tiêu bài học

Producer là điểm bắt đầu của phần lớn luồng dữ liệu Kafka. Nếu Producer thiết kế sai Message Key, gửi dữ liệu không có schema hoặc xác nhận thành công khi dữ liệu chưa đủ an toàn, các Consumer phía sau rất khó sửa hậu quả.

Trong bài này, chúng ta sử dụng Python để xây dựng một Producer gửi sự kiện đơn hàng vào Topic order-created. Project không dừng ở ví dụ Hello World. Học viên sẽ thực hành serialization, delivery callback, retry, idempotence, batching, compression, cấu hình môi trường và kiểm thử.

Sau bài học, học viên có thể giải thích một record được tạo ra như thế nào, Kafka chọn Partition ra sao và vì sao tốc độ cao không đồng nghĩa với gửi từng message nhanh nhất có thể.

2. Kafka Producer làm gì?

Kafka Producer là Client công bố record vào Topic. Producer nhận Key, Value, Header và Timestamp từ ứng dụng, chuyển dữ liệu thành byte, chọn Partition, gom record thành batch rồi gửi batch đến Broker đang giữ Leader của Partition.

Producer không gửi dữ liệu một cách mù quáng. Khi kết nối đến bootstrap.servers, Client lấy metadata của Cluster để biết Topic có những Partition nào và Broker nào đang giữ Leader. Metadata được làm mới định kỳ hoặc khi cấu trúc Cluster thay đổi.

Trong hệ thống thương mại điện tử, Producer có thể nằm trong Order Service. Sau khi đơn hàng được ghi thành công vào cơ sở dữ liệu, ứng dụng tạo Event order_created. Event này được Consumer của thanh toán, tồn kho, vận chuyển và Data Warehouse sử dụng.

Một quyết định quan trọng là thời điểm công bố Event. Nếu lưu đơn hàng thành công nhưng gửi Kafka thất bại, dữ liệu nghiệp vụ và luồng sự kiện không đồng nhất. Đây là bài toán Dual Write. Những hệ thống yêu cầu độ tin cậy cao thường sử dụng Transactional Outbox cùng Change Data Capture thay vì tự ghi database và Kafka trong hai thao tác độc lập.

3. Chuẩn bị project Python

Cấu trúc project:

kafka-python-producer/
  app/
    __init__.py
    config.py
    models.py
    producer.py
    main.py
  tests/
    test_models.py
  .env.example
  requirements.txt

Tạo môi trường ảo:

python -m venv .venv
source .venv/bin/activate

Trên Windows PowerShell:

.venv\Scripts\Activate.ps1

File requirements.txt:

confluent-kafka
python-dotenv
pydantic
pytest

Cài thư viện:

pip install -r requirements.txt

Trong bài này, chúng ta dùng confluent-kafka, Python Client dựa trên librdkafka. Client này có hiệu suất tốt, hỗ trợ idempotence, transaction và nhiều cấu hình cần thiết cho Production.

Tạo .env.example:

KAFKA_BOOTSTRAP_SERVERS=localhost:9092
KAFKA_ORDER_TOPIC=order-created
KAFKA_CLIENT_ID=order-service

Không đưa file .env chứa thông tin thật lên Git.

4. Producer đầu tiên

Tạo file app/producer.py:

import json
from confluent_kafka import Producer


class OrderProducer:
    def __init__(self, bootstrap_servers: str, client_id: str):
        self._producer = Producer(
            {
                "bootstrap.servers": bootstrap_servers,
                "client.id": client_id,
            }
        )

    def send(self, topic: str, key: str, value: dict) -> None:
        self._producer.produce(
            topic=topic,
            key=key.encode("utf-8"),
            value=json.dumps(value).encode("utf-8"),
        )
        self._producer.flush()

Sử dụng Producer:

from app.producer import OrderProducer


producer = OrderProducer(
    bootstrap_servers="localhost:9092",
    client_id="order-service",
)

event = {
    "event_id": "evt-0001",
    "event_type": "order_created",
    "order_id": "ORD-10001",
    "customer_id": "CUS-001",
    "total_amount": 1250000,
    "currency": "VND",
}

producer.send(
    topic="order-created",
    key=event["order_id"],
    value=event,
)

Đoạn code có thể gửi Event nhưng chưa tối ưu. flush() sau từng record khiến Producer chờ toàn bộ message trong hàng đợi được gửi xong. Cách này làm mất lợi ích batching và giảm throughput nghiêm trọng.

Chúng ta sẽ cải tiến từng phần.

5. Serialization và cấu trúc Event

Kafka chỉ lưu Key và Value dưới dạng byte. Producer cần serializer để chuyển object Python thành byte. Consumer cần deserializer tương ứng để khôi phục dữ liệu.

JSON dễ đọc và thuận tiện khi học, nhưng cần chú ý:

• Quy ước kiểu dữ liệu không chặt chẽ.
• Thay đổi field có thể làm Consumer lỗi.
• Payload thường lớn hơn Avro hoặc Protobuf.
• Số thập phân và thời gian cần quy ước rõ.
• Không nên đưa object Python không thể JSON serialize vào Event.

Tạo model bằng Pydantic:

from datetime import datetime, timezone
from decimal import Decimal
from pydantic import BaseModel, Field


class OrderCreatedEvent(BaseModel):
    event_id: str
    event_type: str = "order_created"
    event_version: int = 1
    event_time: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )
    order_id: str
    customer_id: str
    total_amount: Decimal
    currency: str = "VND"
    status: str = "created"

Model kiểm tra dữ liệu trước khi gửi. Event không hợp lệ sẽ bị chặn tại Producer thay vì lan truyền sang nhiều hệ thống.

Serializer:

def serialize_event(event: OrderCreatedEvent) -> bytes:
    return event.model_dump_json().encode("utf-8")

Trong hệ thống lớn, Schema Registry cùng Avro, Protobuf hoặc JSON Schema giúp quản lý compatibility. Nội dung này sẽ được trình bày trong Bài 8.

6. Message Key và cách chọn Partition

Message Key ảnh hưởng trực tiếp đến thứ tự, phân phối tải và khả năng mở rộng.

Nếu dùng order_id làm Key, toàn bộ Event của cùng đơn hàng thường đi vào cùng Partition. Kafka có thể duy trì thứ tự created, paid, packed, shipped cho từng đơn hàng.

Nếu dùng một Key cố định cho tất cả Event, toàn bộ dữ liệu dồn vào một Partition. Những Partition còn lại gần như không có tải. Đây là Hot Partition.

Nếu không gửi Key, Client phân phối record để cân bằng batch. Throughput có thể tốt nhưng Event của cùng một đơn hàng không chắc nằm cùng Partition.

Một Key tốt cần:

• Gắn với đơn vị cần bảo toàn thứ tự.
• Có đủ độ đa dạng để phân phối dữ liệu.
• Ổn định trong suốt vòng đời đối tượng.
• Không chứa thông tin nhạy cảm nếu không cần thiết.
• Được các Producer cùng Domain sử dụng nhất quán.

Không nên chọn Key chỉ dựa trên mong muốn chia đều tuyệt đối. Cần ưu tiên ranh giới nghiệp vụ cần giữ thứ tự.

7. Gửi dữ liệu bất đồng bộ

Phương thức produce() thường đưa record vào hàng đợi nội bộ rồi trả quyền điều khiển cho ứng dụng. Network Thread chịu trách nhiệm gom batch và gửi dữ liệu.

Producer cải tiến:

from confluent_kafka import KafkaException, Message, Producer


class OrderProducer:
    def __init__(self, config: dict):
        self._producer = Producer(config)

    def send(self, topic: str, key: str, value: bytes) -> None:
        try:
            self._producer.produce(
                topic=topic,
                key=key,
                value=value,
                on_delivery=self._delivery_report,
            )
            self._producer.poll(0)
        except BufferError as exc:
            self._producer.poll(1)
            raise KafkaException(str(exc)) from exc

    @staticmethod
    def _delivery_report(error, message: Message) -> None:
        if error is not None:
            print(f"Delivery failed: {error}")
            return

        print(
            "Delivered "
            f"topic={message.topic()} "
            f"partition={message.partition()} "
            f"offset={message.offset()}"
        )

    def close(self, timeout: float = 10.0) -> None:
        remaining = self._producer.flush(timeout)
        if remaining > 0:
            raise RuntimeError(f"{remaining} message chưa được gửi")

Ứng dụng gọi flush() khi chuẩn bị tắt, không gọi sau mọi record. poll() phục vụ callback và các sự kiện nội bộ của Client.

8. Delivery Callback và xử lý lỗi

Lệnh produce() thành công chỉ cho biết record đã vào hàng đợi cục bộ. Nó chưa chứng minh Broker đã ghi record.

Delivery Callback được gọi khi Kafka xác nhận thành công hoặc khi việc gửi thất bại vĩnh viễn. Callback nên ghi log có cấu trúc với Topic, Partition, Offset, Event ID và nguyên nhân lỗi.

Không nên chỉ print lỗi rồi tiếp tục như không có chuyện gì. Tùy nghiệp vụ, ứng dụng có thể:

• Retry lỗi tạm thời.
• Ghi Event vào Outbox.
• Đưa dữ liệu vào hàng đợi dự phòng.
• Trả lỗi cho request nếu Event là một phần bắt buộc.
• Phát cảnh báo khi tỷ lệ lỗi vượt ngưỡng.

Callback chạy trong ngữ cảnh Client. Không nên thực hiện công việc chậm như gọi API hoặc ghi database đồng bộ bên trong callback vì có thể cản trở quá trình xử lý message khác.

9. acks, retry và độ tin cậy

acks xác định số mức xác nhận cần thiết trước khi record được coi là gửi thành công.

acks=0 không chờ phản hồi từ Broker. Độ trễ thấp nhưng Producer không biết record có được ghi hay không.

acks=1 chờ Leader ghi record. Nếu Leader hỏng trước khi Follower sao chép, dữ liệu có thể mất.

acks=all chờ tất cả In-Sync Replica đáp ứng điều kiện xác nhận. Đây là lựa chọn phù hợp cho dữ liệu nghiệp vụ quan trọng khi kết hợp với Replication và min.insync.replicas thích hợp.

Cấu hình Producer:

config = {
    "bootstrap.servers": "localhost:9092",
    "client.id": "order-service",
    "acks": "all",
    "retries": 10,
    "delivery.timeout.ms": 120000,
    "request.timeout.ms": 30000,
}

Retry không nên vô hạn. Ứng dụng cần giới hạn thời gian gửi bằng delivery.timeout.ms. Khi hết thời gian, callback nhận lỗi và hệ thống quyết định xử lý tiếp.

10. Idempotent Producer

Giả sử Broker đã ghi record nhưng phản hồi bị mất trên mạng. Producer không nhận được xác nhận nên gửi lại. Nếu không có cơ chế chống trùng, Broker có thể lưu hai record giống nhau.

Idempotent Producer gắn thông tin Producer ID và Sequence Number để Broker phát hiện retry trùng trong phiên Producer.

config = {
    "bootstrap.servers": "localhost:9092",
    "client.id": "order-service",
    "enable.idempotence": True,
    "acks": "all",
}

Idempotence không có nghĩa toàn bộ hệ thống tự động Exactly Once. Nó chủ yếu ngăn record trùng do retry của Producer trong các điều kiện được hỗ trợ. Consumer vẫn cần xử lý idempotent và hệ thống đích vẫn cần bảo vệ khỏi ghi trùng.

Nếu ứng dụng cần ghi nhiều record vào nhiều Partition như một đơn vị nguyên tử, có thể sử dụng Kafka Transaction. Transaction sẽ được phân tích sâu trong Bài 6.

11. Batch, linger và compression

Kafka đạt throughput cao nhờ xử lý theo batch.

batch.size kiểm soát kích thước batch mục tiêu cho mỗi Partition. linger.ms cho phép Producer chờ một thời gian ngắn để gom thêm record. compression.type xác định thuật toán nén.

Ví dụ:

config = {
    "bootstrap.servers": "localhost:9092",
    "client.id": "order-service",
    "enable.idempotence": True,
    "linger.ms": 10,
    "batch.size": 65536,
    "compression.type": "zstd",
}

Tăng linger.ms có thể tăng throughput nhưng cũng tăng độ trễ. Batch quá lớn có thể dùng nhiều bộ nhớ. Compression giảm network và storage nhưng tiêu tốn CPU.

Không có cấu hình tối ưu cho mọi hệ thống. Cần benchmark bằng payload thật, throughput thật và mục tiêu latency thật.

12. Thiết kế Event chuẩn doanh nghiệp

Một Event tốt nên có Envelope nhất quán:

{
  "event_id": "evt-0001",
  "event_type": "order_created",
  "event_version": 1,
  "event_time": "2026-09-18T08:30:15Z",
  "source": "order-service",
  "correlation_id": "req-88421",
  "data": {
    "order_id": "ORD-10001",
    "customer_id": "CUS-001",
    "total_amount": 1250000,
    "currency": "VND"
  }
}

event_id giúp truy vết và chống trùng. event_type mô tả điều đã xảy ra. event_version hỗ trợ schema evolution. event_time phản ánh thời điểm nghiệp vụ. source cho biết hệ thống tạo Event. correlation_id liên kết Event với request hoặc chuỗi xử lý.

Không nên đặt password, token, số thẻ đầy đủ hoặc dữ liệu cá nhân không cần thiết trong Event. Kafka có retention và nhiều Consumer, vì vậy một field nhạy cảm có thể bị sao chép đến nhiều hệ thống.

13. Bảo mật thông tin kết nối

Không ghi username và password trực tiếp trong source code. Cấu hình nên được đọc từ environment variable hoặc Secret Manager.

import os
from dotenv import load_dotenv


load_dotenv()

producer_config = {
    "bootstrap.servers": os.environ["KAFKA_BOOTSTRAP_SERVERS"],
    "client.id": os.getenv("KAFKA_CLIENT_ID", "order-service"),
}

Production thường bật TLS và SASL. Secret phải được xoay vòng, giới hạn quyền và không xuất hiện trong log. Producer chỉ nên có quyền ghi vào những Topic cần thiết.

14. Logging, metrics và kiểm thử

Mỗi lần gửi lỗi nên ghi:

• Event ID.
• Topic.
• Message Key.
• Loại lỗi.
• Số lần retry.
• Thời gian gửi.

Metrics quan trọng gồm số record gửi thành công, tỷ lệ lỗi, request latency, retry rate, queue size, batch size và compression ratio.

Unit test nên tập trung vào model, serializer, Key và logic tạo Event. Integration test khởi động Kafka bằng Docker, gửi record rồi dùng Consumer xác nhận Value, Key và Header.

Không nên mock mọi thứ. Producer có thể vượt unit test nhưng vẫn lỗi vì advertised listener, quyền ACL, giới hạn kích thước message hoặc schema không tương thích.

15. Project Producer đơn hàng hoàn chỉnh

File app/main.py:

import os
import uuid
from decimal import Decimal

from app.models import OrderCreatedEvent
from app.producer import OrderProducer


def main() -> None:
    config = {
        "bootstrap.servers": os.getenv(
            "KAFKA_BOOTSTRAP_SERVERS", "localhost:9092"
        ),
        "client.id": "order-service",
        "enable.idempotence": True,
        "acks": "all",
        "linger.ms": 10,
        "compression.type": "zstd",
    }

    producer = OrderProducer(config)

    try:
        for index in range(1, 101):
            event = OrderCreatedEvent(
                event_id=str(uuid.uuid4()),
                order_id=f"ORD-{index:05d}",
                customer_id=f"CUS-{index % 20:04d}",
                total_amount=Decimal("1250000"),
            )

            producer.send(
                topic="order-created",
                key=event.order_id,
                value=event.model_dump_json().encode("utf-8"),
            )
    finally:
        producer.close()


if __name__ == "__main__":
    main()

Chạy chương trình:

python -m app.main

Dùng Console Consumer hoặc Kafka UI kiểm tra Key, Partition và Offset.

16. Những lỗi thường gặp

Gọi flush sau mỗi record

Cách này làm Producer gần như đồng bộ, batch nhỏ và throughput thấp. Chỉ flush khi cần đóng ứng dụng hoặc tại ranh giới nghiệp vụ đặc biệt.

Không kiểm tra Delivery Callback

Record vào local queue chưa đồng nghĩa Broker đã ghi thành công.

Dùng Key cố định

Toàn bộ Event dồn vào một Partition và tạo Hot Partition.

Không có Event ID

Consumer khó truy vết và chống xử lý trùng.

Gửi payload quá lớn

Kafka phù hợp với Event, không phù hợp để nhúng file dung lượng lớn. Nên lưu file trong Object Storage và gửi URI cùng metadata.

Ghi bí mật vào source code

Credential dễ bị đưa lên Git hoặc lộ trong log. Hãy dùng Environment Variable hoặc Secret Manager.

17. Tổng kết

Kafka Producer chịu trách nhiệm nhiều hơn việc gửi một chuỗi JSON. Producer quyết định cách serialize dữ liệu, cách chọn Partition, mức xác nhận, chiến lược retry, cơ chế chống trùng và hiệu quả batching.

Các điểm cần nhớ:

• Kafka lưu Key và Value dưới dạng byte.
• Message Key quyết định ranh giới thứ tự và phân phối Partition.
• produce() thường hoạt động bất đồng bộ.
• Delivery Callback mới cho biết kết quả gửi cuối cùng.
• acks=all tăng độ tin cậy khi Cluster được cấu hình đúng.
• Idempotent Producer hạn chế record trùng do retry.
• Batch và compression là nền tảng throughput cao của Kafka.
• Event cần ID, type, version, timestamp và source rõ ràng.

Bài tiếp theo sẽ xây dựng Kafka Consumer bằng Python, quản lý Offset, Consumer Group, Rebalance và Consumer Lag.

18. Câu hỏi ôn tập

1. produce() trả về thành công có chứng minh Broker đã ghi record không?
2. Vì sao Message Key ảnh hưởng đến thứ tự dữ liệu?
3. Điều gì xảy ra nếu mọi record dùng cùng một Key?
4. acks=all bảo vệ dữ liệu như thế nào?
5. Idempotent Producer giải quyết trường hợp nào?
6. Vì sao không nên gọi flush() sau từng record?
7. linger.ms tạo ra đánh đổi gì?
8. Event Envelope nên chứa những field nào?

19. Tài liệu tham khảo

1. Apache Kafka Producer Configurations: https://kafka.apache.org/documentation/#producerconfigs
2. Apache Kafka Design: https://kafka.apache.org/documentation/#design
3. Apache Kafka Protocol and Message Format: https://kafka.apache.org/documentation/#implementation
4. Confluent Kafka Python Client: https://docs.confluent.io/kafka-clients/python/current/overview.html
5. Apache Kafka Security: https://kafka.apache.org/documentation/#security