Chương 3: Kafka Topic Design & Schema Events

Chương 3: Kafka Topic Design & Schema Events#

Thành viên phụ trách: Võ Thành Minh Tuệ

1. Bối cảnh và Mục tiêu#

Trong task này, nhóm tập trung triển khai hạ tầng Message Broker bằng nền tảng Apache Kafka, đóng vai trò là “trạm trung chuyển” dữ liệu trung tâm để hấp thụ các sự kiện đồ thị được phát ra từ Parser Service. Hệ thống yêu cầu năng lực phân luồng sự kiện vào các topic chuyên biệt (dành cho node, edge, metadata, error). Về khía cạnh định dạng, mỗi gói tin (message) bắt buộc phải đóng gói kèm theo các trường quản lý phiên bản (schema_version) và dấu thời gian thực thi (event_time) nhằm đảm bảo tính toàn vẹn của dữ liệu luồng.

2. Cách tiếp cận và Lý do lựa chọn#

Bài toán kiến trúc đầu tiên nhóm đặt ra khi thiết kế hệ thống tin nhắn là: nên tích hợp toàn bộ các loại sự kiện vào chung 1 topic duy nhất để dễ quản lý, hay phân tách chúng thành các topic riêng lẻ?

Nhóm đã đưa ra quyết định phân mảnh luồng dữ liệu thành 4 topic độc lập:

  1. cpg_node_events: kênh chuyên chở siêu dữ liệu của các node cú pháp AST/CPG.

  2. cpg_edge_events: kênh chuyên chở định tuyến của các cạnh liên kết (CFG, DFG).

  3. cpg_metadata_events: kênh quản lý siêu dữ liệu tập tin (kích thước vật lý, độ dài dòng, mã băm nội dung).

  4. Error_parser: hoạt động với vai trò là hàng đợi sự cố (Dead Letter Queue - DLQ), thu gom các gói tin xử lý lỗi hoặc vi phạm định dạng schema.

Cơ sở của việc phân rã cấu trúc:

  • Tối ưu hóa tài nguyên tiêu thụ (Decoupled Consumers): ứng dụng Neo4j Connector chỉ có nhu cầu đăng ký lắng nghe (subscribe) 2 topic cấu trúc là cpg_node_events và cpg_edge_events để dựng hình đồ thị. Ở chiều ngược lại, Spark Streaming Job chỉ cần tương tác duy nhất với cpg_metadata_events. Việc chia tách topic giúp triệt tiêu hoàn toàn chi phí tài nguyên phát sinh từ việc các hệ thống tiêu dùng (consumers) phải đọc và lọc bỏ những gói tin ngoại lai không thuộc miền xử lý của chúng.

  • Khả năng tương thích ngược Schema (Schema Evolution Security): mọi sự kiện JSON được đính kèm mặc định trường "schema_version": "1.0" và "event_time": "<ISO-8601>". Kiến trúc này cung cấp nền tảng linh hoạt, đảm bảo nếu hệ thống phát hành bản nâng cấp payload 2.0 trong tương lai, các worker cũ vẫn có thể nhận diện phiên bản để phân tích mà không gây ra hiện tượng sập hệ thống (crash).

Cấu trúc JSON Schema của từng loại sự kiện#

Theo yêu cầu đặc tả, mỗi message đẩy lên Kafka bắt buộc phải chứa trường schema_version và event_time. Dưới đây là ví dụ payload thực tế của từng topic:

Topic cpg_node_events - mang siêu dữ liệu của một node cú pháp AST:

{
	"column": null,
	"end_line": null,
	"event_time": "2026-07-24T09:16:04.143142+00:00",
	"event_type": "node",
	"file_path": "transformers/benchmark/benches/llama.py",
	"label": "File",
	"line": null,
	"name": "transformers/benchmark/benches/llama.py",
	"node_id": "node_2c27aae2eed1d6fa",
	"properties": {
		"path": "transformers/benchmark/benches/llama.py"
	},
	"schema_version": "1.0"
}

Topic cpg_edge_events - mang quan hệ định tuyến giữa hai node:

{
	"edge_id": "edge_e08ecfa5e681db67",
	"event_time": "2026-07-23T15:58:47.825503+00:00",
	"event_type": "edge",
	"file_path": "transformers/benchmark/benches/llama.py",
	"properties": {
		"alias": null,
		"module": "os"
	},
	"schema_version": "1.0",
	"source_id": "node_2c27aae2eed1d6fa",
	"target_id": "node_2b37ca837c658a8d",
	"type": "IMPORTS"
}

Topic cpg_metadata_events - mang siêu dữ liệu cấp file:

{
	"content_hash": "75cdf73a78cc1f4e8e21cf0aefe037c5e4c09ae2875b81e1eab8fb60f6358fbf",
	"edge_count": 1668,
	"event_time": "2026-07-24T09:16:04.143142+00:00",
	"event_type": "metadata",
	"file_path": "transformers/benchmark/benches/llama.py",
	"line_count": 354,
	"node_count": 473,
	"parser_version": "ast-basic-v1",
	"schema_version": "1.0",
	"size_bytes": 14722
}

Topic Error_parser - Dead Letter Queue (DLQ) thu gom file có lỗi cú pháp:

{
	"error_type": "SyntaxError",
	"event_time": "2026-07-25T04:30:22.206563+00:00",
	"event_type": "parser_error",
	"file_path": "transformers/test_invalid.py",
	"message": "invalid syntax (transformers/test_invalid.py, line 1)",
	"schema_version": "1.0"
}

Điểm đặc biệt của Error topic: Mọi lỗi được cô lập theo file - 1 file lỗi không làm dừng toàn bộ pipeline. Parser vẫn tiếp tục xử lý các file tiếp theo và đẩy event lỗi vào Error_parser topic để môi trường giám sát theo dõi sau.

3. Mã nguồn & Kết quả thực thi#

Khối lệnh dưới đây trong module kafka/producer.py đảm nhiệm việc tương tác với Kafka AdminClient để tự động cung cấp (provision) các topic nếu broker chưa tồn tại:

from confluent_kafka.admin import AdminClient, NewTopic

FILE_TOPIC_MAP = {
    "node_events.jsonl": "cpg_node_events",
    "edge_events.jsonl": "cpg_edge_events",
    "metadata_events.jsonl": "cpg_metadata_events",
    "parser_error_events.jsonl": "Error_parser"
}

def create_topics_if_not_exists():
    admin_client = AdminClient({'bootstrap.servers': 'localhost:9092'})
    existing_topics = admin_client.list_topics().topics
    
    new_topics = [
        NewTopic(topic, num_partitions=1, replication_factor=1)
        for topic in set(FILE_TOPIC_MAP.values())
        if topic not in existing_topics
    ]
    if new_topics:
        admin_client.create_topics(new_topics)

Lệnh thực thi đẩy luồng sự kiện trực tiếp lên Kafka cluster:

python parser/main.py parse-all --output-target kafka

Output thực tế (Terminal)

Parsed 4611 file(s), skipped 0, errors 0, wrote 2029594 node event(s) and 6062224 edge event(s).

4. Minh chứng thực tế#

Minh chứng 1: Giám sát Topic qua giao diện Kafka UI (http://localhost:8080)#

Danh sách Kafka Topics

Hình 3.1: Dashboard của Kafka UI phản ánh trạng thái hoạt động ổn định của cụm broker, ghi nhận đầy đủ 4 topic đã được cấp phát tự động cùng với lượng tin nhắn đang được lưu trữ.

Minh chứng 2: Kiểm tra chéo định dạng Payload trong topic cpg_edge_events#

Mẫu tin nhắn Kafka Metadata

Hình 3.2: Chi tiết nội dung một Message dạng JSON trong topic cpg_edge_events, thể hiện cấu trúc cạnh liên kết với các trường source_id, target_id, type và các trường bắt buộc như schema_version, event_time.

Minh chứng 3: Bắt lỗi cú pháp và đẩy sự kiện vào topic Error_parser#

Chạy lệnh sau để mô phỏng lỗi cú pháp (parser 1 file lỗi trong thư mục transformers):

python -m parser.main parse-file transformers/test_invalid.py --output-target kafka

Output thực tế (Terminal)

Parsed 1 file(s), skipped 0, errors 1, wrote 0 node event(s) and 0 edge event(s).

Sự kiện lỗi trong topic Error_parser

Hình 3.3: Giao diện Kafka UI hiển thị sự kiện lỗi được sinh ra khi parser phát hiện lỗi cú pháp (SyntaxError tại dòng 1 file test_invalid.py), minh chứng cho việc hệ thống Dead Letter Queue (topic Error_parser) đang bắt và lưu trữ lỗi chính xác.

5. Tự nhận xét và Đúc kết#

Trong quá trình khởi tạo mạng lưới Kafka trên nền tảng Docker (đặc biệt đối với môi trường host Windows), nhóm đã gặp phải một cấu hình mạng khá hóc búa. Mặc dù lệnh docker compose ps báo cáo trạng thái container Kafka là healthy, đoạn mã Python chạy trên máy host liên tục văng ngoại lệ Broker not available.

Qua quá trình debug, nhóm phát hiện nguyên nhân xuất phát từ việc cấu hình bridge network chưa phân lập rõ ràng giao thức kết nối nội bộ và ngoại vi. Vấn đề được giải quyết triệt để sau khi nhóm cấu hình lại tham số KAFKA_ADVERTISED_LISTENERS trong file docker-compose.yml, định tuyến tách biệt cổng nội bộ Docker (29092) và cổng expose ra ngoài cho host (9092). Sự điều chỉnh này đã khắc phục hoàn toàn lỗi kết nối, giúp dữ liệu sự kiện bắt đầu streaming mượt mà lên giao diện Kafka UI.

6. Hướng dẫn chạy lại#

  1. Triển khai cụm Broker Kafka và Kafka UI thông qua Docker Compose:

    docker compose up -d kafka kafka-ui

  2. Khởi chạy luồng đẩy sự kiện từ máy khách (client) lên cụm Kafka:

    python parser/main.py parse-all --output-target kafka

  3. Theo dõi trực tiếp luồng tin nhắn đang truyền tải trên Kafka UI tại: http://localhost:8080

  4. Chạy lệnh sau me kích hoạt topic Error_parser và kiểm tra kết quả nhận lỗi:

    python -m parser.main parse-file transformers/test_invalid.py --output-target kafka
Contents