Data SynchronizationData Synchronization

đơn hàng trong bảng nguồnorders in the source table

12M

đơn hàng trong bảng nguồnorders in the source table

thay đổi lúc cao điểmchanges per second at peak

~3.000/s

thay đổi lúc cao điểmchanges per second at peak

độ trễ từ commit tới searchlag from commit to search

p99 < 2s

độ trễ từ commit tới searchlag from commit to search

dòng lệch sau mỗi lần đối soátdrifted rows after each reconciliation

0

dòng lệch sau mỗi lần đối soátdrifted rows after each reconciliation

Debezium đọc WAL của Postgres qua replication slot và đẩy mỗi thay đổi vào Kafka với key là id đơn hàng. Mỗi target có sink riêng, upsert theo version. Job đối soát so checksum từng khoảng id giữa nguồn và Elasticsearch. Chọn “Connector ngừng một giờ” hoặc “Đối soát phát hiện lệch” để xem cách hệ thống tự phục hồi.Debezium reads the Postgres WAL through a replication slot and publishes every change to Kafka keyed by order id. Each target has its own sink that upserts by version. A reconciliation job compares checksums per id range between the source and Elasticsearch. Pick “Connector down for an hour” or “Reconciliation finds drift” to see how the system recovers.

1. 1

   Polling theo updated_at bỏ sót dòngPolling by updated_at misses rows

   Transaction bắt đầu lúc 10:00:00 và commit lúc 10:00:05 mang updated_at = 10:00:00; poller đã qua watermark 10:00:03 sẽ không bao giờ thấy nó. Phải lùi watermark một khoảng an toàn (ví dụ 5 phút) và upsert idempotent, và polling vẫn không thấy hard delete. CDC đọc WAL theo thứ tự commit nên không gặp vấn đề này.A transaction that starts at 10:00:00 and commits at 10:00:05 carries updated_at = 10:00:00; a poller already past watermark 10:00:03 never sees it. You must lag the watermark by a safety window (say 5 minutes) and upsert idempotently, and polling still cannot see hard deletes. CDC reads the WAL in commit order, so it avoids this.

   Pipeline
2. 2

   Thứ tự theo từng dòngOrdering per row

   Kafka chỉ giữ thứ tự trong một partition, nên key phải là khoá chính. Sink vẫn phải so version (LSN hoặc cột version) vì retry và resync có thể đưa bản cũ tới sau bản mới.Kafka keeps order only within a partition, so the key must be the primary key. The sink must still compare versions (the LSN or a version column) because retries and resyncs can deliver an old image after a newer one.

   Publish-Subscribe
3. 3

   Xử lý delete và tombstoneDeletes and tombstones

   Debezium phát sự kiện op=d rồi một tombstone (value null) để log compaction xoá key. Sink xoá theo sự kiện op=d và bỏ qua tombstone. Elasticsearch chỉ nhớ version của tài liệu đã xoá trong index.gc_deletes (60 giây); muốn chặn bản update cũ “hồi sinh” tài liệu thì dùng soft delete trong index.Debezium emits an op=d event and then a tombstone (null value) so log compaction can drop the key. The sink deletes on op=d and ignores the tombstone. Elasticsearch remembers a deleted document's version only for index.gc_deletes (60 seconds); to stop a late update from resurrecting it, soft-delete inside the index.

   Event Sourcing
4. 4

   Lộ schema nội bộ ra ngoàiLeaking the internal schema

   CDC trên bảng gốc biến từng cột thành hợp đồng với mọi consumer. Với dữ liệu chia sẻ giữa các team, dùng bảng outbox và Outbox Event Router của Debezium: service tự quyết định hình dạng sự kiện, CDC chỉ vận chuyển.CDC on the raw table turns every column into a contract with every consumer. For data shared across teams, use an outbox table with Debezium's Outbox Event Router: the service decides the event shape, CDC only carries it.

   Transactional OutboxAdapter
5. 5

   Đổi schema nguồnSource schema changes

   Dùng Schema Registry với Avro hoặc Protobuf ở chế độ BACKWARD: thêm cột nullable là an toàn, đổi tên hoặc đổi kiểu cột thì bị từ chối. Đổi tên làm theo expand/contract: thêm cột mới, ghi cả hai, chuyển consumer, rồi mới bỏ cột cũ.Use a Schema Registry with Avro or Protobuf in BACKWARD mode: adding a nullable column is safe, renaming or retyping is rejected. Renames follow expand/contract: add the new column, write both, move consumers, then drop the old one.

   Adapter
6. 6

   Backfill khi thêm target mớiBackfill for a new target

   Target mới cần toàn bộ 12M dòng cũ trong khi stream vẫn chạy. Incremental snapshot của Debezium (thuật toán watermark kiểu DBLog) đọc từng chunk theo khoá chính, xen kẽ với sự kiện stream, không khoá bảng và không cần dừng connector.A new target needs all 12M existing rows while streaming continues. Debezium's incremental snapshot (DBLog-style watermarks) reads chunks by primary key, interleaved with streamed events, without locking the table or stopping the connector.

   PipelineStrategy

Sink đọc sự kiện Debezium và upsert vào Elasticsearch với external version, nên sự kiện trùng hoặc đến muộn bị bỏ qua. Job đối soát so digest từng khoảng id và chỉ đồng bộ lại phần lệch.The sink reads Debezium events and upserts into Elasticsearch with an external version, so duplicates and late events are dropped. The reconciliation job compares a digest per id range and resyncs only what differs.

@Component
@RequiredArgsConstructor
public class OrderSearchSink {
    private final ElasticsearchClient es;

    @KafkaListener(topics = "pg.public.orders", groupId = "orders-search")
    public void onChange(ConsumerRecord<String, ChangeEvent> rec) throws IOException {
        ChangeEvent e = rec.value();
        if (e == null) return;                    // tombstone, only for compaction1
        String id = rec.key();
        long version = e.source().lsn();          // grows with every commit2
        try {
            if (e.op() == Op.DELETE) {
                es.delete(d -> d.index("orders").id(id)
                        .version(version).versionType(VersionType.External));3
            } else {                              // c, u, r (snapshot read)
                es.index(i -> i.index("orders").id(id)
                        .version(version).versionType(VersionType.External)
                        .document(OrderDoc.from(e.after())));
            }
        } catch (ElasticsearchException ex) {
            if (ex.status() != 409) throw ex;     // 409: stale or duplicate, skip4
        }
    }
}

type Digest struct{ Count, VSum int64 } // count(*), sum(version)1

type Side interface {
	Digest(ctx context.Context, lo, hi int64) (Digest, error)
	Versions(ctx context.Context, lo, hi int64) (map[int64]int64, error)
}

type Repairer interface {
	Resync(ctx context.Context, ids []int64) error // Debezium signal
	Delete(ctx context.Context, ids []int64) error // straight to target
}

type Reconciler struct {
	src, dst Side
	fix      Repairer
	step     int64 // 10_000 ids per bucket
}

func (r *Reconciler) Run(ctx context.Context, maxID int64) error {
	for lo := int64(0); lo <= maxID && ctx.Err() == nil; lo += r.step {2
		a, err := r.src.Digest(ctx, lo, lo+r.step)
		if err != nil {
			return err
		}
		b, err := r.dst.Digest(ctx, lo, lo+r.step)
		if err != nil {
			return err
		}
		if a != b {3
			if err := r.repair(ctx, lo, lo+r.step); err != nil {
				return fmt.Errorf("bucket %d: %w", lo/r.step, err)
			}
		}
	}
	return ctx.Err()
}

func (r *Reconciler) repair(ctx context.Context, lo, hi int64) error {
	want, err := r.src.Versions(ctx, lo, hi)
	if err != nil {
		return err
	}
	got, err := r.dst.Versions(ctx, lo, hi)
	if err != nil {
		return err
	}
	var stale, orphan []int64
	for id, v := range want {
		if got[id] != v { // missing or different version4
			stale = append(stale, id)
		}
	}
	for id := range got {
		if _, ok := want[id]; !ok {
			orphan = append(orphan, id)
		}
	}
	return errors.Join(r.fix.Resync(ctx, stale), r.fix.Delete(ctx, orphan))
}

Mỗi pattern mở bài học tương ứng trong menu Pattern.Each one opens its lesson in the Patterns menu.