Đồng bộ DB → search index

Đây là cái giá của bài 6.11: giờ có hai nơi lưu cùng một sản phẩm. PostgreSQL là source of truth, index products là bản sao. Bài này trả lời: làm sao bản sao theo kịp, không sai, không mất.

1. Bài toán

• Admin sửa giá, đổi tên, ẩn sản phẩm. Inventory đổi tồn kho hàng nghìn lần mỗi phút trong flash sale.
• Search phải phản ánh trong vài giây, không bao giờ "kẹt" ở giá cũ mãi mãi.
• Thỉnh thoảng cần đổi mapping (thêm analyzer) mà không tắt search.

2. Vấn đề: dual write

@Transactional
public void updatePrice(String id, BigDecimal price) {
  Product p = repo.findById(id).orElseThrow();
  p.setPrice(price);
  repo.save(p);
  searchClient.index(toDoc(p));   // ghi ES ngay trong transaction
}

Trông ổn, nhưng có ít nhất 4 cách hỏng:

1. ES lỗi/timeout → exception → rollback DB. Search sập kéo theo admin không sửa được giá.
2. Bắt exception cho qua → DB 149k, ES 199k mãi mãi.
3. ES ghi thành công, sau đó commit DB lỗi → ES có giá chưa từng tồn tại.
4. Hai request song song: A đặt 149k, B đặt 129k. DB commit theo thứ tự A rồi B, nhưng gọi ES theo thứ tự B rồi A → ES kết thúc ở 149k, sai.

Không có transaction chung giữa PostgreSQL và ES. Đây chính là bài toán dual write đã gặp với Kafka ở bài 14.7.

3. Giải pháp

Nguyên tắc: chỉ ghi DB trong transaction. Việc cập nhật index đi sau, bất đồng bộ, lấy thay đổi từ chính DB. Hai cách:

• Outbox + consumer: cùng transaction ghi thêm một dòng vào bảng outbox; relay đẩy sang Kafka; indexer đọc Kafka và ghi ES.
• CDC với Debezium: đọc WAL (logical replication, bài 6.2) của bảng products, đẩy mọi INSERT/UPDATE/DELETE sang Kafka. Không cần sửa code ghi.

Cách hoạt động của outbox, relay và Debezium đã ở bài 14.7. Ở đây tập trung vào phần riêng của search: indexer phải idempotent, đúng thứ tự, xử lý xóa, và có đường reindex toàn bộ.

4. Code từng bước

Bước 1: Debezium connector (Kafka Connect), chỉ lấy bảng cần thiết:

{
  "name": "shop-products-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "${file:/secrets/db.properties:host}",
    "database.user": "${file:/secrets/db.properties:user}",
    "database.password": "${file:/secrets/db.properties:password}",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "table.include.list": "public.products,public.product_stock",
    "slot.name": "search_indexer",
    "tombstones.on.delete": "true"
  }
}

Topic shop.public.products, key = id sản phẩm → mọi thay đổi của cùng sản phẩm vào cùng partition, giữ thứ tự (bài 8.6).

Bước 2: thêm cột version vào DB để ES biết bản nào mới hơn:

ALTER TABLE products ADD COLUMN version bigint NOT NULL DEFAULT 0;
-- JPA: @Version Long version;  → mỗi UPDATE tăng version

Bước 3: indexer idempotent với external versioning:

@Component
@RequiredArgsConstructor
@Slf4j
public class ProductIndexer {
  private final ElasticsearchClient es;
  private final ProductReadRepository reads;   // đọc lại từ DB/replica

  @KafkaListener(topics = "shop.public.products", groupId = "search-indexer")
  public void onChange(ConsumerRecord<String, ProductChange> rec) throws IOException {
    String id = rec.key();
    if (rec.value() == null || rec.value().op() == Op.DELETE) {   // tombstone hoặc delete
      deleteDoc(id, rec.value() == null ? null : rec.value().before().version());
      return;
    }
    ProductRow row = rec.value().after();
    ProductDoc doc = reads.loadDocument(id);   // join category, brand, stock
    if (doc == null) return;                   // đã bị xóa sau đó, bỏ qua
    try {
      es.index(i -> i.index("products")        // ghi qua alias
          .id(id)
          .version(doc.version())
          .versionType(VersionType.External)   // chỉ ghi nếu version > version đang có
          .document(doc));
    } catch (ElasticsearchException e) {
      if (e.status() == 409) {                 // bản cũ tới muộn: bỏ qua là đúng
        log.debug("stale event for {} v{}", id, doc.version());
        return;
      }
      throw e;   // lỗi khác: để Kafka retry / DLT
    }
  }

  private void deleteDoc(String id, Long version) throws IOException {
    es.delete(d -> {
      d.index("products").id(id);
      if (version != null) d.version(version + 1).versionType(VersionType.External);
      return d;
    });   // 404 cũng coi là thành công
  }
}

• Idempotent: nhận lại cùng event (retry, rebalance) → cùng version → 409 → bỏ qua. An toàn với at-least-once.
• Chống đảo thứ tự: event cũ tới sau bị ES từ chối.
• Đọc lại bản đầy đủ từ DB thay vì tin hoàn toàn vào payload: document search thường join nhiều bảng.
• Ưu tiên soft delete (active=false) ở DB → indexer xóa khỏi index. Hard delete thì dựa vào delete event + tombstone của Debezium; topic compact cũng cần tombstone để xóa key.
• Trong flash sale, stock đổi liên tục: gộp (debounce) theo id trong 1–2 giây rồi dùng _bulk, hoặc chỉ index trạng thái inStock chứ không index số lượng chính xác.

Bước 4: reindex toàn bộ không downtime bằng alias:

# 1. Tạo index mới với mapping mới
curl -X PUT "$SEARCH_URL/products_v2" -H 'Content-Type: application/json' -d @products-v2.json
# 2. Đổ dữ liệu: job đọc DB theo keyset (id > ?), _bulk vào products_v2, refresh_interval -1 cho nhanh
#    Indexer realtime vẫn ghi vào alias "products" (đang trỏ v1) — và ghi thêm vào v2 trong lúc build
# 3. Bật lại refresh, kiểm tra: count, vài query mẫu
# 4. Đổi alias nguyên tử
curl -X POST "$SEARCH_URL/_aliases" -H 'Content-Type: application/json' -d '{
  "actions": [
    { "remove": { "index": "products_v1", "alias": "products" } },
    { "add":    { "index": "products_v2", "alias": "products", "is_write_index": true } }
  ]}'
# 5. Giữ v1 vài ngày để rollback, rồi xóa

Vì dùng external version, event realtime và bulk load ghi chồng nhau cũng không sai: bản nào version cao hơn thắng. Có thể thay bước 2 bằng: ghi lại offset Kafka, bulk load, rồi cho consumer group mới chạy từ offset đó vào v2.

Bước 5: test bằng curl:

curl -X PATCH localhost:8080/api/admin/products/p-1001 -H 'Content-Type: application/json' \
     -d '{"price":149000}'
sleep 3
curl -s "$SEARCH_URL/products/_doc/p-1001" -u "$SEARCH_USER:$SEARCH_PASSWORD" | jq '._version, ._source.price'
# 8
# 149000

5. Cách hoạt động

 Admin API          PostgreSQL (source of truth)
    │ UPDATE price      │
    └──────────────────►│ WAL ──► logical slot "search_indexer"
                        │                │
                        │          Debezium (Kafka Connect)
                        │                ▼
                        │      Kafka: shop.public.products   key = productId
                        │                │   (cùng key → cùng partition → đúng thứ tự)
                        │                ▼
                        │◄─── đọc lại ── Indexer (consumer group search-indexer)
                        │                │  index(id, version, External)
                        │                ▼
                        │      alias "products" ──► products_v2
                        │                │  refresh_interval 1s
   User search ─────────┼───────────────►│  (thấy giá mới sau ~1–3 giây)
   Checkout ───────────►│ đọc giá từ DB, không từ ES

Near real-time: ES ghi vào buffer, chỉ sau refresh (mặc định 1 giây, chỉ trên index có search gần đây) thì document mới tìm được. Tổng độ trễ = Debezium + Kafka + indexer + refresh, thường 1–5 giây. Tăng refresh_interval lên 5–30 giây giúp ghi nhanh hơn nhiều khi tải lớn. Không gọi refresh=true sau mỗi lần ghi trên production.

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Outbox: đã có outbox cho event domain; muốn event có ý nghĩa nghiệp vụ (ProductPriceChanged).
• CDC: nhiều nguồn ghi vào bảng (admin, import, script SQL), không muốn sửa code ghi.
• Cron reindex: dữ liệu ít thay đổi, chấp nhận trễ (catalog tài liệu nội bộ).
• Dù chọn cách nào, luôn có job full reindex chạy được bất cứ lúc nào: đó là lưới an toàn cho bug indexer.

Bạn đừng hứa "ngay lập tức". Hãy đưa ra một con số cam kết và thiết kế để độ lệch không gây hại:

1. Đặt SLO cho độ trễ: "99% thay đổi xuất hiện trên search trong 5 giây". Đo được thì mới nói chuyện được.
2. Chỗ quan trọng đọc DB: trang chi tiết, giỏ hàng, checkout luôn lấy giá và tồn kho từ PostgreSQL. Search chỉ để tìm. Khách thấy 199k ở list, vào chi tiết thấy 149k: rẻ hơn thì vui, đắt hơn thì checkout vẫn đúng giá thật.
3. Read-your-writes cho admin: sau khi admin lưu, UI admin hiển thị bản từ DB, hoặc indexer nhận cờ để ghi với refresh=wait_for riêng cho thao tác admin (ít, nên chấp nhận được).
4. Flash sale: hết hàng thì search có thể còn hiện vài giây; nút "Mua" kiểm tra tồn kho thật nên không bán quá.

Câu ngắn gọn cho PO: "Search là bản sao để tìm nhanh, trễ tối đa vài giây. Giá và tồn kho khách trả tiền luôn lấy từ nguồn chính." Đây là eventual consistency có kiểm soát (bài 14.5).

Đúng, đồng bộ bất đồng bộ mà không đo thì sẽ hỏng âm thầm. Theo dõi 4 tầng:

Timer lag = Timer.builder("search.index.lag")
    .publishPercentiles(0.5, 0.99).register(meterRegistry);
// trong indexer, sau khi index thành công:
lag.record(Duration.between(doc.updatedAt(), Instant.now()));

Thêm một job đối soát mỗi đêm: lấy mẫu ngẫu nhiên 1.000 id, so version giữa DB và ES; lệch thì index lại và báo cáo. Nó bắt được bug mà metric lag không thấy (ví dụ indexer bỏ qua nhầm event).

Q1. Vì sao không ghi DB rồi gọi Elasticsearch ngay trong service?

🇻🇳 Tiếng Việt:

Đó là dual write, không có transaction chung. ES lỗi thì hoặc rollback DB, hoặc bỏ qua làm hai bên lệch vĩnh viễn. Hai request song song còn có thể ghi ES sai thứ tự. Em chỉ ghi DB, rồi đồng bộ bất đồng bộ qua outbox hoặc CDC Debezium vào Kafka, indexer đọc và ghi ES.

🇬🇧 English:

That is a dual write with no shared transaction. If Elasticsearch fails, we either roll back the database or ignore it and the two stay inconsistent forever. Concurrent requests can also reach Elasticsearch out of order. I only write to the database and sync asynchronously through an outbox or Debezium CDC into Kafka, and an indexer consumes and writes to Elasticsearch.

Q2. Làm sao indexer an toàn khi event bị lặp hoặc đến sai thứ tự?

🇻🇳 Tiếng Việt:

Key Kafka là id sản phẩm để cùng sản phẩm vào cùng partition. Mỗi dòng có version tăng dần, indexer ghi với version_type external, nên ES chỉ nhận bản có version lớn hơn. Event lặp hoặc cũ trả về 409 và được bỏ qua. Xóa thì dùng soft delete hoặc delete event và tombstone.

🇬🇧 English:

The Kafka key is the product id, so changes for one product go to the same partition. Each row has an increasing version and the indexer writes with external version type, so Elasticsearch only accepts a higher version. Duplicate or stale events get a 409 and are skipped. Deletes use soft delete, or delete events with tombstones.

Q3. Đổi mapping mà không downtime thế nào?

🇻🇳 Tiếng Việt:

App luôn đọc ghi qua alias. Em tạo index mới với mapping mới, bulk load từ DB, đồng thời cho event realtime ghi vào index mới. Kiểm tra số lượng và query mẫu, rồi đổi alias nguyên tử trong một lệnh _aliases. Giữ index cũ vài ngày để rollback.

🇬🇧 English:

The app always reads and writes through an alias. I create a new index with the new mapping, bulk load it from the database, and also send realtime events to the new index. After checking counts and sample queries, I swap the alias atomically in one _aliases call, and keep the old index for a few days for rollback.

Cách nhớ ngắn gọn

• Không dual write: chỉ ghi DB, đồng bộ qua outbox/CDC → Kafka → indexer.
• Key = id, version external: lặp và đảo thứ tự đều vô hại.
• Reindex: index mới + alias swap.
• Eventual vài giây: search để tìm, checkout đọc DB.
• Đo lag mọi tầng + đối soát hằng đêm.