Series Apache Kafka · Bài 5/10. Một khách bấm tạo đơn trên website. API kiểm tra dữ liệu, phát event vào Kafka và trả biên nhận. Consumer đọc event rồi lưu PostgreSQL. Bài này ghép các phần đã học thành một hệ thống local có thể quan sát từ trình duyệt đến SQL, đồng thời giải quyết tình huống consumer đọc lại một event đã ghi database.
Đây là luồng tiếp nhận đơn để học event processing, chưa phải cổng thanh toán hoặc hệ thống đặt hàng production. Kafka local từ bài 2 vẫn dùng được. Python chạy trên host trong môi trường bài 3. Chúng ta dùng Flask, confluent-kafka, Psycopg 3 và PostgreSQL 17. Không cần sửa topic của bài 4.
Mục lục
- Phân biệt tiếp nhận và xử lý
- Chuẩn bị PostgreSQL và topic
- Schema và nguyên tắc chống trùng
- Website phát sinh đơn
- Consumer ghi PostgreSQL
- Chạy toàn bộ lab
- Kiểm tra khi đọc lại
- Lỗi và giới hạn kiến trúc
- Bài tập
- Tài liệu tham khảo
1. Đã tiếp nhận khác với đã xử lý
Trong lab, HTTP 202 chỉ được trả sau khi producer nhận callback xác nhận gửi thành công. Nó có nghĩa yêu cầu đã được Kafka tiếp nhận theo cấu hình producer, chưa có nghĩa consumer đã ghi PostgreSQL. Nếu consumer đang dừng, website vẫn có thể nhận đơn trong khi Kafka giữ event chờ xử lý.
Mỗi yêu cầu có request_id do phía gọi cung cấp và dùng lại khi retry cùng một yêu cầu. Lab sử dụng request_id làm event_id và order_id để dễ quan sát idempotency. Trong hệ thống thật, ba định danh này có thể tách riêng; cần xác định rõ phạm vi duy nhất và cách xử lý cùng ID nhưng payload khác nhau. Bài này sẽ từ chối payload xung đột ở consumer.
2. Chuẩn bị database, thư viện và topic
Tạo thư mục mới kafka-orders-app, kích hoạt môi trường ảo rồi cài thư viện. Có thể dùng lại venv bài 3. Cổng PostgreSQL được map sang 5433 để giảm khả năng trùng với database đã chạy trên máy. Mật khẩu dưới đây chỉ là thông tin của database lab local.
python -m pip install "Flask>=3.1,<4" "psycopg[binary]>=3.2,<4" "confluent-kafka==2.16.1"
docker run -d --name kafka5-pg -e POSTGRES_USER=lab -e POSTGRES_PASSWORD=labpass -e POSTGRES_DB=orderslab -p 127.0.0.1:5433:5432 -v kafka5_pgdata:/var/lib/postgresql/data postgres:17Nếu container tên kafka5-pg đã tồn tại, kiểm tra nó thay vì chạy lại docker run. Chờ database sẵn sàng rồi tạo topic riêng:
docker exec kafka5-pg pg_isready -U lab -d orderslab
docker exec kafka-lab /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic web_orders_v1 --partitions 3 --replication-factor 1Named volume giữ database qua việc tạo lại container. Kafka và PostgreSQL đều chạy trong container, nhưng hai chương trình Python chạy trên host nên dùng localhost:9092 và localhost:5433. Không dùng tên container như hostname từ host nếu chưa có cơ chế DNS phù hợp.
3. Thiết kế bảng để đọc lại không tạo đơn trùng
Tạo file schema.sql rồi chạy lệnh nhập bên dưới. processed_events giữ dấu event đã xử lý và payload để kiểm tra xung đột. orders là kết quả nghiệp vụ. Khóa chính event_id ngăn cùng event được đánh dấu hai lần; khóa chính order_id ngăn hai đơn cùng ID. Hai thay đổi phải nằm trong một transaction.
CREATE TABLE IF NOT EXISTS processed_events (
event_id text PRIMARY KEY,
payload jsonb NOT NULL,
processed_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS orders (
order_id text PRIMARY KEY,
event_id text NOT NULL UNIQUE REFERENCES processed_events(event_id),
customer_name text NOT NULL,
amount_vnd bigint NOT NULL CHECK (amount_vnd > 0),
created_at timestamptz NOT NULL,
stored_at timestamptz NOT NULL DEFAULT now()
);docker exec -i kafka5-pg psql -U lab -d orderslab < schema.sql
Nếu chỉ ghi processed_events rồi commit riêng, chương trình có thể chết trước khi ghi orders. Khi đọc lại, nó tưởng event đã hoàn tất và bỏ qua, gây thiếu đơn. Transaction buộc cả hai cùng thành công hoặc cùng rollback. Kafka offset là một commit khác ở hệ thống khác, nên ta sắp xếp database trước, offset sau và xử lý duplicate ở database.
4. Website tạo đơn và phát event
Tạo app.py. Trang chủ cung cấp form đơn giản. Trường request_id được tạo khi mở trang; cùng lần submit hoặc cùng lần retry cần giữ ID và dữ liệu. Mở lại trang chủ tạo request mới. Giá trị amount_vnd được kiểm tra là số nguyên dương trong giới hạn lab.
import json
import uuid
from datetime import datetime, timezone
from flask import Flask, request, jsonify
from confluent_kafka import Producer
app = Flask(__name__)
@app.get("/")
def home():
request_id = str(uuid.uuid4())
timestamp = datetime.now(timezone.utc).isoformat()
return f'''<!doctype html><html lang="vi"><meta charset="utf-8">
<title>Kafka Orders Lab</title><h1>Tạo đơn hàng</h1>
<form method="post" action="/orders">
<input type="hidden" name="occurred_at" value="{timestamp}">
<label>Request ID <input name="request_id" value="{request_id}" required></label><br>
<label>Tên khách <input name="customer_name" value="Nguyễn An" required></label><br>
<label>Số tiền VND <input name="amount_vnd" type="number" value="850000" min="1" required></label><br>
<button>Tiếp nhận đơn</button></form></html>'''
@app.post("/orders")
def create_order():
data = request.get_json(silent=True) if request.is_json else request.form
if not data:
return jsonify(error="Missing request data"), 400
try:
event_id = str(uuid.UUID(str(data.get("request_id", ""))))
name = str(data.get("customer_name", "")).strip()
raw = data.get("amount_vnd")
if isinstance(raw, bool) or not str(raw).isdigit():
raise ValueError("amount_vnd must be an integer")
amount = int(raw)
if not name or len(name) > 100 or not 1 <= amount <= 10**12:
raise ValueError("Invalid name or amount")
except (ValueError, TypeError) as exc:
return jsonify(error=str(exc)), 400
occurred_at = data.get("occurred_at")
try:
datetime.fromisoformat(occurred_at)
except (ValueError, TypeError):
return jsonify(error="Invalid occurred_at"), 400
event = {"event_id": event_id, "event_type": "order.created",
"order_id": event_id, "customer_name": name,
"amount_vnd": amount, "occurred_at": occurred_at}
results = []
producer = Producer({"bootstrap.servers": "localhost:9092",
"enable.idempotence": True, "acks": "all",
"delivery.timeout.ms": 10000})
try:
producer.produce("web_orders_v1", key=event_id.encode(),
value=json.dumps(event, ensure_ascii=False).encode("utf-8"),
on_delivery=lambda err, msg: results.append(err))
pending = producer.flush(12)
except Exception:
app.logger.exception("Kafka send failed")
return jsonify(error="Cannot confirm acceptance; retry same request_id and payload"), 503
if pending or not results or results[0] is not None:
return jsonify(error="Cannot confirm acceptance; retry same request_id and payload"), 503
return jsonify(status="accepted", order_id=event_id), 202
if __name__ == "__main__":
app.run(host="127.0.0.1", port=5000, debug=False)Form tạo occurred_at một lần khi mở trang và gửi trong trường hidden. Retry cùng yêu cầu phải giữ nguyên request_id, occurred_at, customer_name và amount_vnd. Client JSON phải cung cấp occurred_at dạng ISO có múi giờ. Production cần ghi nhận request và timestamp vào kho bền vững để các lần retry trả cùng kết quả.
Producer được tạo riêng theo request để ví dụ callback dễ hiểu; tạo client và flush đồng bộ mỗi request không tối ưu throughput. Production thường tái sử dụng client cùng cơ chế callback, timeout và xử lý đồng thời phù hợp. Flask development server chỉ chạy local; không dùng nó để triển khai công khai.
5. Consumer lưu database rồi commit offset
Tạo sink.py. DSN mặc định trỏ database lab; có thể đổi bằng biến môi trường PG_DSN. Với mỗi record, Psycopg 3 mở connection context, commit transaction khi khối with kết thúc thành công và rollback nếu có exception. Chỉ sau khối đó consumer mới commit Kafka offset.
import json
import os
from datetime import datetime
import psycopg
from psycopg.types.json import Jsonb
from confluent_kafka import Consumer, KafkaException
DSN = os.getenv("PG_DSN", "postgresql://lab:labpass@localhost:5433/orderslab")
c = Consumer({"bootstrap.servers": "localhost:9092",
"group.id": os.getenv("GROUP_ID", "orders-pg-v1"),
"auto.offset.reset": "earliest", "enable.auto.commit": False})
c.subscribe(["web_orders_v1"])
try:
while True:
msg = c.poll(1.0)
if msg is None:
continue
if msg.error():
raise KafkaException(msg.error())
event = json.loads(msg.value().decode("utf-8"))
if event.get("event_type") != "order.created":
raise ValueError("Unsupported event type")
if type(event.get("amount_vnd")) is not int or event["amount_vnd"] <= 0:
raise ValueError("Invalid amount")
created_at = datetime.fromisoformat(event["occurred_at"])
if created_at.tzinfo is None:
raise ValueError("Timestamp must include timezone")
with psycopg.connect(DSN) as conn:
with conn.cursor() as cur:
cur.execute("INSERT INTO processed_events(event_id,payload) VALUES (%s,%s) ON CONFLICT DO NOTHING RETURNING event_id",
(event["event_id"], Jsonb(event)))
inserted = cur.fetchone()
if inserted:
cur.execute("INSERT INTO orders(order_id,event_id,customer_name,amount_vnd,created_at) VALUES (%s,%s,%s,%s,%s)",
(event["order_id"], event["event_id"], event["customer_name"], event["amount_vnd"], created_at))
else:
cur.execute("SELECT payload FROM processed_events WHERE event_id=%s", (event["event_id"],))
if cur.fetchone()[0] != event:
raise ValueError("Same event_id with conflicting payload")
c.commit(message=msg, asynchronous=False)
print("STORED" if inserted else "DUPLICATE", event["order_id"], flush=True)
except KeyboardInterrupt:
print("Stopping")
finally:
c.close()Tham số SQL được truyền bằng %s và tuple, không ghép tên khách vào chuỗi SQL. INSERT ... ON CONFLICT DO NOTHING RETURNING cho biết event mới hay đã có. Nếu đã có, consumer đọc payload cũ để phát hiện cùng ID nhưng khác nội dung. Payload trùng thì bỏ qua insert orders; payload khác thì dừng để điều tra.
Khi database lỗi, exception thoát khối with và rollback. Kafka offset chưa được commit nên consumer có thể đọc lại sau khi sự cố được sửa. Nếu database đã commit nhưng Kafka commit lỗi, record cũng có thể đọc lại; khóa event_id và transaction giúp lần đọc lại không tạo thêm đơn. Đây là idempotent sink trong phạm vi các bảng đã thiết kế, không phải transaction phân tán giữa Kafka và PostgreSQL.
6. Chạy toàn bộ hệ thống
python sink.py
# Terminal khác
python app.pyMở http://127.0.0.1:5000 trên máy của bạn, nhập tên và số tiền rồi submit. API trả JSON accepted với HTTP 202; terminal sink phải in STORED và order_id tương ứng. Truy vấn database để kiểm tra bằng bằng chứng bền vững, không chỉ nhìn log:
docker exec kafka5-pg psql -U lab -d orderslab -c "SELECT order_id, customer_name, amount_vnd, created_at FROM orders ORDER BY stored_at DESC LIMIT 10;"
docker exec kafka5-pg psql -U lab -d orderslab -c "SELECT count(*) AS orders, coalesce(sum(amount_vnd),0) AS total_vnd FROM orders;"Nếu tạo ba yêu cầu mới 850.000 đồng mỗi lần, kỳ vọng có ba đơn và tổng 2.550.000 đồng trên database lab ban đầu rỗng. Nếu submit lại cùng request_id và payload, API có thể tiếp nhận thêm bản ghi Kafka nhưng consumer nhận ra duplicate và số đơn không tăng. Tạo request_id mới cho giao dịch mới.
7. Kiểm tra đọc lại và consumer tạm dừng
Dừng sink bằng Ctrl+C, tạo thêm một đơn trên website. API vẫn có thể trả accepted nếu Kafka đang chạy. Query PostgreSQL chưa thấy đơn mới là điều phù hợp vì consumer chưa xử lý. Mở lại sink cùng group và đợi STORED; kiểm tra SQL lần nữa.
Để replay toàn bộ topic mà không xóa offset cũ, dừng sink rồi chạy group mới. Database giữ processed_events, nên những event đã có chỉ in DUPLICATE.
GROUP_ID=orders-pg-replay-v1 python sink.pyĐối chiếu count và sum trước sau replay. Nếu tăng số đơn dù event_id giống nhau, hãy kiểm tra unique constraint, phạm vi transaction và việc có đang dùng cùng database không. Replay không tự khôi phục dữ liệu đã bị Kafka retention loại bỏ. Bảng processed_events cũng cần chính sách lưu đủ lâu tương ứng yêu cầu replay.
8. Lỗi và giới hạn cần hiểu
Nếu API trả 503 do timeout, kết quả có thể chưa xác định: broker có thể đã nhận nhưng phản hồi bị mất. Retry cùng request_id và payload giúp downstream nhận diện trùng. Không tạo ID mới chỉ vì request timeout. API lab chưa lưu idempotency state bền vững; hệ thống thật cần lưu request và trả lại cùng kết quả cho lần gọi lặp.
Kiến trúc này phát event trước rồi consumer mới tạo bản ghi đơn. Nếu ứng dụng thật cần đồng thời lưu đơn nguồn và phát event, ghi database rồi produce là bài toán dual write có cửa sổ lỗi. Transactional outbox và CDC là cách thường dùng để nối hai bước: ghi đơn cùng outbox trong một transaction database, sau đó chuyển outbox sang Kafka. Bài 6 giới thiệu CDC với Debezium để hiểu cơ chế lấy thay đổi từ database.
Schema lab xử lý order.created một lần trên mỗi order_id, không giải quyết cập nhật trạng thái nhiều phiên bản. Email, trừ kho hay gọi cổng thanh toán không nằm trong transaction PostgreSQL này; mỗi tác dụng phụ cần chiến lược idempotency riêng. Với lỗi record độc hại, consumer hiện dừng; bài 8 sẽ bổ sung retry và dead letter topic.
9. Bài tập có hướng dẫn kiểm tra
Bài 1: gửi lại cùng request_id và payload hai lần. Kỳ vọng Kafka có thể chứa hai record nhưng orders chỉ có một dòng. Đối chiếu cả processed_events và orders.
Bài 2: giữ request_id nhưng đổi amount_vnd. Consumer phải phát hiện conflicting payload và không commit record lỗi. Sửa nguồn hoặc dùng topic lab mới để làm lại; không bỏ qua xung đột bằng cách đổi constraint.
Bài 3: thêm channel vào form, event và bảng. Gợi ý: tạo migration ALTER TABLE, cập nhật API và sink cùng hợp đồng schema; đừng chỉ sửa giao diện rồi mong database tự có cột.
Bạn hoàn thành lab khi quan sát được đơn từ browser đến Kafka và PostgreSQL, giải thích được HTTP 202, transaction và thứ tự commit, đồng thời chứng minh replay không tăng số đơn. Bài tiếp theo sẽ đưa thay đổi MySQL tự động vào Kafka bằng Kafka Connect và Debezium.
Tài liệu tham khảo
- Flask Documentation — request và local development server.
- Psycopg 3 Transactions — commit, rollback và connection context.
- PostgreSQL INSERT — ON CONFLICT và RETURNING.
- Confluent Python Client — producer callback và consumer commit.
@techdata.ai · Series Apache Kafka