8 lab RabbitMQ và Kafka

Lab 01 · RabbitMQ routing

Goal · Mục tiêu

Tạo direct/topic/fanout topology, mandatory publish và alternate exchange. Kiểm chứng route đúng, unroutable được quan sát và redeclare sai attributes làm channel fail.

Mô hình cần chứng minh. Routing quyết định queue nào nhận message; confirm quyết định kết quả publish ở phía broker. Hai câu hỏi này phải có hai oracle riêng. Message được alternate exchange route thành công không còn là unroutable đối với mandatory. [R1] [R2] [R11]

Fixture và cấu hình

Dùng bốn message cố định: M1=order.created, M2=order.paid, M3=stock.created, M4=no.route. Gửi đúng một lần mỗi message vào từng exchange trong subcase riêng, bật confirms và mandatory=true. Giữ consumer tắt đến khi đã ghi hết confirm/return để tách routing khỏi processing.

Việc phải làm

1. Khai báo exchange/queue/binding bằng cấu hình versioned; dùng durable queue, không auto-delete. Export cấu hình, chạy lại cùng declaration và xác nhận không tạo tài nguyên mới.
2. Chạy ba subcase direct/topic/fanout từ state sạch. Lưu publish sequence → event ID, confirm, return và routing key; sau đó drain vào ledger theo (queue, event_id) và so set với bảng oracle.
3. Tạo exchange thử AE riêng: gửi M4 khi AE có binding, rồi xóa binding của AE và gửi M4 trong subcase sạch khác. Không dùng metric aggregate để suy ra có return hay không; phải bắt callback/protocol event của chính message đó.
4. Mở channel dùng riêng cho negative control. Khai báo lại một queue đã tồn tại với durable trái ngược; sau lỗi, mở channel mới và khai báo đúng để kiểm tra recovery. [R3]

Failure injection · Chủ động gây lỗi

Acceptance / evidence · 8 tiêu chí nghiệm thu

Review quyết định. Giải thích khi nào dùng return để producer xử lý trực tiếp, khi nào dùng AE để giữ message cho vận hành. Cả hai đều cần ownership và cảnh báo; “không bị return” không chứng minh consumer đã hoàn thành.

Lab 02 · Confirm, ack và crash window

Goal · Mục tiêu

Manual ack + publisher confirm. Crash consumer sau DB commit trước ack; chứng minh redelivery nhưng chỉ một side effect nhờ unique idempotency key.

Mô hình cần chứng minh. Đặt business update và inbox marker trong cùng DB transaction, rồi mới ack delivery. Sau crash giữa commit và ack, message có thể được giao lại nhưng business update không được lặp. Confirm và consumer ack độc lập; quorum queue cần đa số replica để tiếp tục tiến triển. [R2] [R4]

Fixture và cấu hình

Fixture chuẩn có E001…E100, mỗi event tăng counter đúng 1. Inbox có unique key (handler_id,event_id); counter và marker commit nguyên tử. Dùng một consumer, manual ack, prefetch 1 cho các ca crash; publisher lưu trạng thái PENDING bền vững trước send, rồi cập nhật kết quả confirm. Các ca HA dùng quorum queue ba replica và tạm tắt consumer để giữ backlog đã confirm.

on_delivery(event):
    BEGIN DB TRANSACTION
    inserted = INSERT inbox(handler_id, event_id, payload_hash) IF ABSENT
    if inserted: APPLY business_effect(+1)
    else: ASSERT stored_payload_hash == event.payload_hash
    COMMIT DB TRANSACTION
    HIT_HOOK("AFTER_DB_COMMIT_BEFORE_ACK", event.id)
    ACK(delivery_on_its_original_channel)

Việc phải làm

1. Chạy baseline 100 event. Trước khi bật consumer, chụp đủ 100 confirm cùng queue state; sau khi drain, đối chiếu 100 ID với inbox và counter. Không chỉ kiểm số row inbox.
2. Đặt hook E050 sau DB commit nhưng trước ack; test controller đọc DB bằng connection khác, xác nhận side effect đã có và ack chưa gửi, rồi kill consumer. Restart và giữ nguyên DB. Chạy thêm ca E025 trước commit: rollback phải được chứng minh trước khi recovery.
3. Tạo các negative control riêng: gửi hai bản cùng E005 khi không có unique constraint; và ack E001 trước khi DB commit rồi crash. Lưu kết quả sai, reset, bật thiết kế đúng và chạy lại; không coi negative control là cấu hình ứng dụng đạt yêu cầu.
4. Tạo ca một event U001: publisher đã nhận confirm callback nhưng bị giữ trước khi ghi kết quả vào durable publish ledger. Kill publisher, kiểm ledger vẫn PENDING, restart và gửi lại cùng event ID. Đây là outcome không rõ với publisher sau phục hồi, không phải bằng chứng đã mô phỏng mất packet confirm trên mạng.
5. Với fixture HA riêng, xác nhận ba replica online và backlog 100 event đã confirm; dừng leader rồi đọc hết dữ liệu. Sau reset, dừng hai replica, chờ trạng thái mất đa số, mới gửi 10 ID mới; phục hồi replica và đối soát/retry pending.

Failure injection · Chủ động gây lỗi

Acceptance / evidence · 8 tiêu chí nghiệm thu

Review quyết định. Ghi rõ idempotency bảo vệ handler nào và trong bao lâu. Unique constraint phải bao trùm transaction nghiệp vụ; check-then-insert không có ràng buộc nguyên tử là chưa đủ. Ack phải gắn đúng channel/delivery; không “khôi phục” bằng ack delivery tag cũ trên channel mới.

Lab 03 · Prefetch và backpressure

Goal · Mục tiêu

Benchmark prefetch 1/20/200 với consumer nhanh/chậm. Đo ready/unacked, memory, fairness, throughput; kích hoạt downstream chậm và chọn giới hạn.

Mô hình cần chứng minh. Prefetch giới hạn số delivery chưa ack, không tự giới hạn mọi buffer trong ứng dụng hoặc tốc độ producer. Lab phải đo cả broker backlog, bộ nhớ process và công việc đang giữ ở client. Không mặc định prefetch lớn hơn sẽ tốt hơn. [R5] [R6]

Fixture và cấu hình

Mỗi ô benchmark dùng E0001…E6000, serialized payload được pad đến đúng 1.024 byte/message, cùng TLS, persistence, confirms và topology. Hai consumer, mỗi consumer một channel và một worker xử lý tuần tự, manual ack; không bật single-active-consumer. Prefetch per consumer lần lượt 1 / 20 / 200. Profile F: cả hai downstream stub xử lý 5 ms; profile S: C1 xử lý 5 ms, C2 xử lý 50 ms. Timer của stub là workload có chủ đích, không dùng để đoán thời điểm kill.

Warm-up 200 event trong namespace riêng, không trộn vào số liệu. Prefill 6.000 event đã confirm; đăng ký đủ hai consumer và giữ worker tại barrier để cả hai có backlog trước khi bắt đầu đo. Có 3 prefetch × 2 profile × 3 lần chạy = 18 hàng benchmark, không phải 18 tiêu chí nghiệm thu.

throughput = 6000 / (last_effect_time - worker_release_time)
fairness_J = (n_C1 + n_C2)^2 / (2 * (n_C1^2 + n_C2^2))
backlog_growth ≈ max(0, ingress_rate - sustainable_processing_rate)
catch_up_time ≈ backlog / (recovery_processing_rate - ongoing_ingress_rate)

Công thức catch-up chỉ có ý nghĩa khi tốc độ xử lý lúc recovery lớn hơn ingress. Fairness J đo phân bố số việc giữa hai consumer, không chứng minh latency tốt; profile consumer chậm không nên bị ép nhận 50% công việc.

Việc phải làm

1. Chốt trước resource budget, p95 processing-age budget và drain deadline. Ghi rõ queue type, policy, payload bytes, thời gian downstream stub và định nghĩa clock đo age. Thu raw metric mỗi 2 giây cùng log mỗi delivery/effect.
2. Chạy đủ ma trận 18 hàng. Với mỗi hàng, lưu ready/unacked, client admitted/in-flight, application buffer bytes, RSS, broker memory, số completion mỗi consumer, throughput, p50/p95 end-to-end age và thời gian drain. Không điền số liệu ước lượng vào cột actual.
3. Chạy subcase saturation riêng: chặn cả hai downstream ngay từ đầu, cho broker giao message đến hạn mức rồi chụp state. Với 2 consumer và prefetch P, kiểm broker unacked không vượt 2×P; giữ gate tới khi đã có snapshot, sau đó release.
4. Chạy subcase producer liên tục, tối đa 6.000 ID và watchdog 300 giây. Khi downstream bị giữ tại barrier, bật admission control: application buffer tối đa 2 MiB payload, số công việc tối đa 2×P, producer outstanding-confirm tối đa 100. Ghi quyết định pause/throttle/reject hoặc lưu pending bền vững; không drop thầm.
5. Gỡ lỗi, drain toàn bộ event đã được chấp nhận và tính recovery capacity. Chọn cấu hình trên kết quả median/range của ba lần chạy cùng các budget; lưu cả kết quả không đạt thay vì chỉ giữ cấu hình thắng.

Failure injection · Chủ động gây lỗi

Acceptance / evidence · 8 tiêu chí nghiệm thu

Review quyết định. Một mẫu RSS thấp chưa chứng minh bộ nhớ bị chặn. Phân biệt broker credit, buffer của client library, worker queue và publisher buffer; đo tất cả phần có thể tăng. Giữ cùng durability/security profile khi so hiệu năng.

Lab 04 · Retry/DLX operations

Goal · Mục tiêu

Tách transient/permanent error, retry có TTL và bounded attempts, DLQ metadata. Viết replay CLI có dry-run, rate limit và audit; tái hiện dead-letter loop rồi sửa.

Mô hình cần chứng minh. Retry phải phân loại lỗi và có điểm dừng; redelivery do crash không được tự biến thành một lần retry nghiệp vụ mới. DLX không mặc nhiên là chuyển giao không mất dữ liệu: profile an toàn ở đây dùng quorum source queue, dead-letter-strategy=at-least-once và overflow=reject-publish. TTL là điều kiện hết hạn, không phải bộ hẹn giờ chính xác. [R4] [R7] [R8]

Fixture và cấu hình

Baseline không có transport fault: 31 business attempts, 14 effects, 5 event quarantine và 12 retry transitions. Hai queue retry có queue-level TTL lần lượt 1.000 ms và 2.000 ms, không có consumer; DLX route trở lại work queue. Không trộn TTL khác nhau trong cùng retry queue. Lưu application attempt và quyết định theo (event_id,attempt); x-death dùng để chẩn đoán, không thay business retry counter.

DLQ collector lưu raw payload + metadata vào quarantine store bền vững rồi mới ack. Replay CLI đọc snapshot từ store này, không consume/requeue DLQ để giả lập dry-run. Mỗi entry chứa event/attempt ID, payload hash, error class, first/last seen, origin exchange/routing key, schema version, death history và owner.

Việc phải làm

1. Tạo work/retry/DLQ topology và export policy thực tế. Trong error handler, giữ nguyên event ID, tăng business attempt chỉ khi chuyển sang lần thử kế tiếp. Republish retry/DLQ bằng confirms + mandatory routing check, rồi mới ack delivery gốc; publish fail/unknown thì không xác nhận mất dữ liệu.
2. Chạy fixture baseline để đạt đúng oracle 31/14/5/12. Ghi thời điểm vào/ra mỗi retry queue; kiểm không xử lý trước TTL trong cùng clock đo đã chốt, có tolerance đo clock, và kết thúc trong deadline. Không assert đúng chính xác 1 hoặc 2 giây.
3. Dừng tại hook sau retry publish confirm, trước ack gốc rồi kill worker. Kiểm transport duplicate nhưng decision ledger và idempotency giữ đúng tập side effect; attempt nghiệp vụ tối đa 3 vẫn giữ. Tái hiện thêm retry queue không route được về work bằng cách gỡ binding DLX rồi khôi phục.
4. Tái hiện loop trong topology cách ly: loop.work → consumer reject(requeue=false) → DLX → loop.delay(TTL) → loop.work. Mỗi vòng có consumer rejection; controller dừng sau 5 lần rejection hoặc watchdog. Sau đó sửa bằng bounded attempts và terminal DLQ; không dựa vào cơ chế phát hiện chu kỳ không có rejection của broker. [R7]
5. Viết replay CLI có hai chế độ dry-run/execute, filter theo snapshot ID, allowlist schema, giới hạn số event, throttle 2 event/giây với burst=1, checkpoint và audit. Nâng cấp handler cho P, sửa downstream cho X; không sửa payload hoặc đổi event ID để né dedupe.
6. Dry-run năm entry quarantine, execute, rồi kill CLI ở hook sau publish confirm trước audit checkpoint và chạy lại. Lưu duplicate attempt do replay; chỉ đánh dấu hoàn tất khi có bằng chứng downstream đã xử lý hoặc đã đối soát, không coi confirm là business success.

Failure injection · Chủ động gây lỗi

Acceptance / evidence · 8 tiêu chí nghiệm thu

Review quyết định. Nêu chi phí giữ dead letters ở source, nguy cơ retry amplification và cách cảnh báo DLQ age. At-least-once transfer vẫn có thể duplicate; replay phải dùng lại idempotency contract của live traffic, không phải đường tắt bỏ kiểm tra.

Lab 05 · Kafka partition và lag

Goal · Mục tiêu

Chọn key accountId, tạo hot key, đo lag per partition. Tăng partitions và ghi rõ ảnh hưởng mapping/order; scale consumers vượt partition count.

Mô hình cần chứng minh. Lag cần nhìn theo partition và theo tuổi dữ liệu, không chỉ cộng thành một số. Tăng partition không di chuyển record cũ và có thể đổi partition của cùng key cho record mới. Consumer group thông thường không tăng số partition được xử lý song song chỉ bằng cách thêm member vượt partition count. [K1] [K2]

Fixture và cấu hình

Dùng topic ba partition và key accountId. Để oracle độc lập ngôn ngữ, lab định nghĩa partitioner riêng: partition = số ở cuối accountId mod P; ví dụ account-0003 vào p0 khi P=3, vào p3 khi P=6. Ghi rõ đây là partitioner của fixture, không phải thuật toán mặc định của mọi Kafka client.

Các topic mới, không có transaction/control record trong fixture B/H/X, không compaction và không hết retention trong lúc chạy. Khi consumer chưa chạy, committed offset được khởi tạo 0; lưu cả log start/end và checkpoint để diễn giải lag. Trong dữ liệu có transaction hoặc compaction, chênh offset không tự bằng số business event chưa xử lý.

Việc phải làm

1. Chạy B rồi H trong namespace riêng. Lưu producer delivery metadata gồm event ID, key, partition, offset; dừng consumer trước snapshot lag. Với H, chỉ resume p1/p2, giữ p0 bằng barrier để có oracle lag p0=216 và p1=p2=0.
2. Chạy B với group lần lượt 1, 3, 5 member, một topic và assignment strategy đã pin. Đợi group ổn định, export assignment và completions. Báo số member có partition và member idle; không khẳng định throughput tăng tuyến tính.
3. Trong ca X không an toàn, giữ consumer của record cũ, tăng partition lên 6 và đợi producer thấy metadata mới. Gửi 12 record sequence=31; cho p3–p5 chạy trước p0–p2. Quan sát ít nhất account-0003 có sequence 31 được áp dụng trước record cũ.
4. Làm lại X an toàn: ngừng producer, drain và xác nhận đủ 360 effect/checkpoint cũ, thay partition count, chờ metadata mới rồi mới phát 12 record tiếp theo. Ghi kế hoạch rollback không giả định có thể giảm partition count của topic cũ.
5. Với I, dừng một follower, chờ ISR=2, publish 10 ID A. Tiếp tục dừng follower thứ hai, giữ controller quorum còn đa số, chờ ISR=1 rồi thử 10 ID B. Ghi từng response/error/timeout; phục hồi follower, chờ ISR đủ và retry/đối soát B bằng event ID.

Failure injection · Chủ động gây lỗi

Acceptance / evidence · 8 tiêu chí nghiệm thu

Review quyết định. acks=all chờ tập ISR hiện hành, không đồng nghĩa mọi replica được cấu hình luôn online; min.insync.replicas là điều kiện durability của profile ghi này, không được đồng nhất với RabbitMQ quorum. Nêu lựa chọn giữ ordering per account hay chia nhỏ key để giảm hotspot, và chi phí cutover. [K3] [K4]

Lab 06 · Offset và rebalance

Goal · Mục tiêu

Manual commit sau DB transaction. Cho processing vượt max.poll.interval hoặc kill member; quan sát revoke/reassign/duplicate và sửa bằng bounded worker + pause/resume.

Mô hình cần chứng minh. Committed offset là vị trí tiếp theo cần đọc, không phải “message đã nhận từ poll”. Với worker song song, chỉ commit qua dải record liên tục đã hoàn tất. Pause/resume giúp chặn nạp thêm việc, nhưng không thay nghĩa vụ tiếp tục poll. Rebalance cũng không tự rollback DB effect của worker cũ. [K5] [K2]

Fixture và cấu hình

Topic mới có ba partition, mỗi partition 20 record ở offset 0…19, tổng 60 event; không transaction, compaction hoặc retention deletion trong ca thử. Dùng dynamic membership, group.protocol=classic, không đặt group.instance.id, tắt auto commit và đặt max.poll.interval.ms=5000, max.poll.records=10. Ghi rõ client thực sự hỗ trợ các lựa chọn này; client không hỗ trợ phải ghi BLOCKED, không âm thầm đổi protocol.

Baseline xử lý tuần tự và commit sau DB transaction. Bản sửa dùng owner loop gọi consumer API, hai worker, application admission cap 50 record; pause khi admitted≥40, resume khi ≤20, mỗi poll nạp tối đa 10 record. Admitted bao gồm việc đang chạy và đang chờ. Chọn fetch-byte limits và RSS budget riêng vì max.poll.records không giới hạn toàn bộ dữ liệu client đã fetch/cache. [K5]

on_worker_done(partition, offset, generation):
    if generation != current_owner_generation:
        discard_offset_completion()  # DB effect vẫn phải được inbox bảo vệ
    else:
        mark_DB_committed(partition, offset)
        next = first_not_completed_from_current_checkpoint(partition)
        commit_next_offset(partition, next)  # Chỉ owner loop thực hiện

on_partition_revoked(partition):
    stop_admission(partition)
    drain_with_deadline_or_cancel_workers(partition)
    commit_only_contiguous_completed_offsets_if_still_allowed()
    invalidate_generation(partition)

Việc phải làm

1. Tạo inbox + business effect trong một DB transaction; chỉ commit offset sau commit DB thành công. Lưu topic/partition/offset cùng event ID, nhưng dedupe bằng event ID khi đó là định danh nghiệp vụ ổn định.
2. Giữ p1/offset7 sau DB commit trước offset commit, kill member rồi khởi động lại. Chạy ca poll-timeout riêng: giữ owner loop không poll cho đến khi vượt 5 giây và coordinator xác nhận reassignment. Không thay bằng chỉ giữ worker trong khi owner vẫn poll.
3. Chạy ca kill member đột ngột khi có record in-flight; dùng assignment ở member sống và coordinator để chứng minh chuyển quyền. Không đòi callback revoke từ process đã bị kill.
4. Tạo completion gap riêng trên p0: hoàn tất và commit offset 0…6 để checkpoint=7; giữ offset7, cho offset8/9 hoàn tất DB trước. Kiểm checkpoint vẫn bằng 7, rồi release offset7 và cho phép checkpoint tiến tới 10.
5. Bật bounded worker + pause/resume; chặn downstream 20 giây nhưng owner loop vẫn poll. Lưu mọi poll interval và worker-queue size. Thực hiện thêm controlled rebalance khi worker còn dở để kiểm generation fencing, cancellation và contiguous commit.
6. Sau mọi lỗi, resume và drain; đối chiếu 60 ID, business effects, checkpoint cuối, assignment ổn định và không còn worker cũ commit offset sau mất quyền. Đánh dấu rõ DB effect đã commit của worker cũ có thể tồn tại và được inbox dedupe.

Failure injection · Chủ động gây lỗi

Acceptance / evidence · 8 tiêu chí nghiệm thu

Review quyết định. Nêu đánh đổi giữa batch size, throughput, replay khi crash và thời gian rebalance. Tăng max.poll.interval chỉ thay cửa sổ timeout; nó không sửa worker queue vô hạn, commit sớm hay thiếu idempotency.

Lab 07 · Kafka transactions và external DB

Goal · Mục tiêu

Read-process-write Kafka transaction với read_committed; sau đó thêm DB side effect để chứng minh nó ngoài EOS. Áp outbox/inbox hoặc reconciliation.

Mô hình cần chứng minh. Kafka transaction có thể gộp output records và consumer offsets trong Kafka. Nó không tự gộp transaction của database ngoài Kafka. read_committed lọc record của transaction bị abort/chưa commit, không phải bộ lọc duplicate business event cho mọi nguồn dữ liệu. [K6] [K7]

Fixture và cấu hình

Input topic mới có một partition, 10 event E01…E10, mỗi event có value=1; transform tạo một output cùng event ID với value=2. Dùng một processing slot với transactional.id ổn định trong phạm vi run, không trùng slot khác; output checker dùng isolation.level=read_committed. RF/min ISR áp dụng cả các topic nghiệp vụ; cấu hình transaction/offset internal topics cũng phải đủ replication cho cluster lab.

initialize_transactional_producer(stable_slot_id)
for input_record:
    begin_Kafka_transaction()
    send(output_record_with_same_event_id)
    wait_send_result()
    HIT_HOOK("AFTER_OUTPUT_SEND_BEFORE_TX_COMMIT", event_id)
    send_next_input_offset_to_transaction(current_group_metadata)
    commit_Kafka_transaction()

Hai checker có group độc lập: một read_committed, một read_uncommitted. Checker không được làm thay đổi group offset của processor. Đếm event visible theo ID, không lấy chênh offset làm số business output vì transaction có thể tạo khoảng trống/control records. [K5]

Việc phải làm

1. Chạy baseline Kafka-only: output và input offset commit trong cùng Kafka transaction. Sau đó giữ E05 ở hook sau send, trước transaction commit; để checker read_uncommitted xác nhận có bản chưa commit rồi kill processor.
2. Restart cùng transactional ID theo slot và đúng recovery API của client. Đợi transaction cũ được giải quyết/abort; processor đọc lại từ committed input offset. Lưu cả record vật lý được read_uncommitted thấy và tập output read_committed cuối cùng.
3. Chạy ca crash sau transaction commit nhưng trước ghi progress log cục bộ. Chạy thêm ca hai producer dùng cùng transactional ID: khởi tạo producer thứ hai rồi buộc producer cũ tiếp tục thao tác để kiểm fencing; không để hai slot hợp lệ chia sẻ ID.
4. Thêm DB counter ngoài Kafka transaction, cố ý chưa có unique inbox. Với E05, commit DB rồi kill trước Kafka transaction commit. Recovery xử lý lại input; đo output Kafka vẫn đúng 10 trong khi counter DB=11. Đây là negative control có chủ đích.
5. Sửa bằng DB transaction chứa inbox + business update + outbox; commit input offset chỉ sau DB commit. Relay publish outbox sang Kafka; sink dùng inbox theo event ID. Kiểm cả crash trước commit DB, sau commit DB trước offset, và relay sau publish success trước đánh dấu outbox đã gửi.
6. Viết reconciliation so tập input, inbox, business effects, outbox, output receipts và sink effects. Relay có thể gửi lại cùng event ID; downstream không được cộng lại. Ghi rõ đây là atomicity tại DB cộng eventual delivery, không phải distributed transaction tự động giữa DB và Kafka.

Failure injection · Chủ động gây lỗi

Acceptance / evidence · 8 tiêu chí nghiệm thu

Review quyết định. So sánh chi phí transaction latency, fencing, outbox cleanup và reconciliation. Dedupe record phải sống đủ lâu cho replay; xóa inbox sớm có thể làm một event cũ tạo side effect mới. Với side effect không giao dịch được như email/payment API, cần idempotency contract của đích hoặc workflow đối soát riêng.

Lab 08 · Schema và incident drill

Goal · Mục tiêu

Test v1/v2 rolling compatibility, inject malformed event vào DLQ, broker/consumer restart và disk/lag alert. Viết runbook chẩn đoán, recovery và safe replay.

Mô hình cần chứng minh. Compatibility là quan hệ giữa writer, reader và cách giải mã thực tế; “thêm field là an toàn” không phải kết luận áp dụng cho mọi format. Recovery chỉ hoàn tất khi schema, quyền truy cập, dữ liệu và checkpoint đều đúng; trạng thái process “running” chưa đủ. [S1] [K8]

Fixture và cấu hình

Chọn JSON có validation hoặc một schema format khác và pin serializer/reader. Fixture Kafka gồm 10 event v1 + 10 event v2, ID không trùng; mỗi event có amountMinor=100. v1 có event_id, accountId, amountMinor; v2 thêm currency. Contract lab: reader v1 bỏ qua field không biết; reader v2 dùng currency mặc định VND khi v1 không có; toàn bộ v2 fixture cũng dùng VND. Đây là contract cần test, không suy từ tên format.

Ba poison records riêng: bytes không parse được, amountMinor="one hundred" trái kiểu số, và schema_version=99 chưa hỗ trợ. Reader phải đọc raw bytes và deserialize trong lớp có thể quarantine; không để lỗi deserializer làm poll mắc kẹt mãi. Poison không có event ID dùng định danh quarantine theo (source topic, partition, offset).

Fixture incident khác có 200 valid event để tạo lag xác định; không trộn với 20 event schema. Dùng lại profile quorum/ISR và security của các lab trước, nhưng mỗi subcase có namespace riêng. Tất cả tải được giới hạn bởi fixture/watchdog.

Việc phải làm

1. Chạy ma trận writer/reader 2×2 trong state sạch: v1→v1, v1→v2, v2→v1, v2→v2. Mỗi ô dùng 10 event đúng writer version và kiểm field/giá trị sau decode. Sau đó rolling upgrade consumer trước producer, lưu version đang chạy và trạng thái xử lý trong suốt rollout.
2. Inject ba poison records xen giữa valid records. Với Kafka, dùng transaction gộp DLQ record và source next offset; với pipeline RabbitMQ tương đương, chỉ ack sau quarantine bền vững hoặc publish confirm đã kiểm routing. Giữ raw bytes/hash, original coordinates, error class và schema fingerprint để replay có chọn lọc.
3. Dừng/restart lần lượt một RabbitMQ quorum leader, một Kafka broker và một consumer; không xóa volume hoặc đổi group ID để che recovery. Đọc lại tập ID đã confirm/ack-success, kiểm dedupe và checkpoint. Tách restart từng thành phần thành subcase để biết nguyên nhân.
4. Tạo lag thật: publish 200 ID, giữ consumer ở barrier, chờ alert lag>100 liên tục 10 giây với scrape 2 giây. Release và drain. Tạo disk alarm RabbitMQ bằng cách nâng ngưỡng disk-free trong sandbox vượt free bytes hiện tại, không lấp đầy ổ hệ điều hành; đọc alarm thật, thử publisher probe, rồi khôi phục ngưỡng trong finally. [R9]
5. Chạy security matrix trên cả hai broker: TLS verification, authentication và least privilege theo resource. Dùng admin chỉ để dựng tài nguyên, account ứng dụng không có admin. Xoay credential, kiểm bằng connection mới để không nhầm session cũ còn sống với credential cũ vẫn hợp lệ.
6. Safe replay 20 valid schema events từ archive/offset range đã chốt vào topic/group cách ly. Một mode dùng lại business ledger để kiểm dedupe; một mode rebuild projection sạch để so hash/kết quả. Không rewind group live, không tự động replay ba poison records chưa có bản sửa.

Failure injection · Chủ động gây lỗi

Acceptance / evidence · 8 tiêu chí nghiệm thu

Review quyết định. RabbitMQ authorization cần đúng vhost/exchange/queue; Kafka cần đúng topic/group và transactional ID nếu dùng transactions. Security probe phải phân biệt TLS failure, authentication failure và authorization deny. Application dùng quyền tối thiểu; replay, monitoring và management access có owner riêng. [R10] [K9]