نقل تغييرات قاعدة البيانات CDC مع Debezium

في مقال Saga ونمط Outbox ذكرنا طريقتين لإخراج الأحداث من جدول ال Outbox: السؤال الدوري Polling، وقراءة سجل المعاملات Transaction Log Tailing، ووعدنا بشرح الثانية مع Debezium بالتفصيل، وهذا هو المقال. والمشكلة التي تحلها أوسع من ال Outbox: عندك قاعدة بيانات PostgreSQL فيها الطلبات والعملاء، وتريد أن يعرف محرك البحث والتقارير وال Cache بكل تغيير فيها لحظة حدوثه، دون أن تعدل كل مكان في الكود يكتب في هذه الجداول، ودون أن تسأل قاعدة البيانات كل ثانية «ما الجديد؟».

وهنا يأتي نقل التغييرات Change Data Capture، واختصاراً CDC: أداة تقرأ السجل الذي تكتب فيه قاعدة البيانات كل تغيير، وتحول كل صف أضيف أو عدل أو حذف إلى حدث في Kafka أو غيرها. والفكرة بسيطة لكن تشغيلها له ثمن قد يملأ قرص قاعدة بياناتك الرئيسية إذا لم تفهمه، وسوف نناقش في هذا المقال ما يلي:

• طرق نقل التغييرات الثلاث، ولماذا قراءة السجل هي الأفضل غالباً.
• كيف يعمل Debezium مع PostgreSQL: ال WAL، والفك المنطقي Logical Decoding، وال Publication، وال Replication Slot، وKafka Connect.
• شكل الحدث، ومعنى REPLICA IDENTITY، واللقطة الأولى Snapshot.
• نمط Outbox مع المحول Outbox Event Router.
• تغيير شكل الجداول Schema Changes وما يحدث للمستهلكين.
• تجربة حقيقية تنتهي بكسر متعمد: المستهلك يتوقف فيكبر ال WAL، ثم يضيع ال Slot، ثم نستعيد.

ثلاث طرق لنقل التغييرات

• السؤال بعمود الوقت: كل دقيقة تسأل WHERE updated_at > :last. بسيطة، لكنها لا ترى الحذف أبداً، وتفوت التغييرات التي لم تحدث العمود، وترى آخر حالة فقط، فإذا تغير الصف مرتين بين سؤالين فاتك التغيير الأول.
• المشغلات Triggers: كل تغيير يكتب صفاً في جدول آخر، وقد شرحناها في مقال سجل التدقيق. ترى كل شيء، لكنها تضيف كتابة على كل معاملة، وما زلت تحتاج من ينقل الجدول إلى الأنظمة الأخرى.
• قراءة السجل Log-Based CDC: قاعدة البيانات تكتب كل تغيير في سجل المعاملات قبل تنفيذه، وهو ال WAL في PostgreSQL، والأداة تقرأ هذا السجل. ترى كل إضافة وتعديل وحذف بترتيب الالتزام Commit، ولا تضيف حملاً على الجداول نفسها، ولا تحتاج أي تعديل في كود التطبيق. وهذه طريقة Debezium.

كيف يعمل Debezium مع PostgreSQL؟

Debezium مشروع مفتوح المصدر برخصة Apache 2.0 بحسب مستودعه، وآخر إصدار عند كتابة المقال 3.7.0.Final الذي صدر في 29 سبتمبر 2026، وتذكر صفحة الإصدار أنه مختبر مع PostgreSQL من 14 إلى 18 ومع Kafka Connect 3.1 وما بعده. والصورة التالية تبين الأجزاء:

1. الفك المنطقي Logical Decoding: ال WAL في أصله سجل للصفحات على القرص، وعندما تضبط wal_level=logical يكتب PostgreSQL معلومات إضافية تسمح بتحويله إلى تغييرات بالصفوف، والقيمة الافتراضية replica لا تكفي، وتغييرها يحتاج إعادة تشغيل.
2. ال Publication: تحدد الجداول التي يرسل تغييراتها البرنامج المدمج pgoutput، وDebezium ينشئها بنفسه إذا أعطيته الصلاحية.
3. ال Replication Slot: هو الطرف الذي يتذكر في PostgreSQL أين وصل Debezium، ويمنع حذف أي جزء من ال WAL لم يؤكد Debezium قراءته، وبسببه لا يضيع أي تغيير إذا توقف Debezium، وبسببه أيضاً يكبر القرص كما سوف ترى.
4. Kafka Connect: Debezium في شكله الأشهر موصل Connector يعمل داخل Kafka Connect، ويحفظ موقعه Offset في موضوع داخلي في Kafka، ويكتب أحداث كل جدول في موضوع باسم prefix.schema.table.

وقد يتساءل البعض: هل أحتاج Kafka حتى أستخدم Debezium؟ والإجابة لا، فهناك Debezium Server، وهو برنامج مستقل يقرأ التغييرات ويرسلها مباشرة إلى وجهات كثيرة منها Redis Streams وNATS JetStream وRabbitMQ وGoogle Pub/Sub وAmazon Kinesis، وهناك Debezium Engine مكتبة داخل تطبيق Java. لكن Kafka Connect يبقى الشكل الأكثر استخداماً، لأنه يعطيك توزيع العمل وحفظ المواقع وموصلات الكتابة Sink Connectors إلى قواعد بيانات ومحركات بحث أخرى، لذلك استخدمناه في التجربة.

شكل الحدث، ومعنى REPLICA IDENTITY

كل حدث فيه مفتاح Key هو المفتاح الأساسي للصف، وقيمة فيها الحقول التالية:

• op: نوع التغيير، r لصف قرئ في اللقطة الأولى، وc للإضافة، وu للتعديل، وd للحذف.
• before وafter: الصف قبل التغيير وبعده.
• source: من أين جاء الحدث، ومنه رقم الموقع في ال WAL lsn ورقم المعاملة txId.

والنقطة التي تفاجئ أغلب الفرق أن before يعتمد على إعداد الجدول REPLICA IDENTITY، والقيمة الافتراضية DEFAULT تجعل PostgreSQL يكتب في ال WAL المفتاح الأساسي فقط للصف القديم، فيصل التعديل دون before، ويصل الحذف ب before فيه المفتاح فقط وباقي الحقول قيم افتراضية، كما سوف ترى في التجربة. فإذا كان المستهلك يحتاج القيمة القديمة، مثل سجل تدقيق أو حساب فرق في مخزون، فاضبط الجدول على REPLICA IDENTITY FULL، وثمن ذلك أن ال WAL يكبر لأن كل تعديل يكتب الصف القديم كاملاً.

اللقطة الأولى Snapshot

ال WAL لا يحتوي البيانات القديمة، فعندما يبدأ Debezium أول مرة يقرأ الجداول كاملة ويرسل كل صف بحدث op=r، ثم ينتقل إلى قراءة التغييرات من النقطة التي بدأت عندها اللقطة، فلا يضيع شيء بين الاثنين. والإعداد snapshot.mode يحدد ذلك، وأهم قيمه بحسب توثيق موصل PostgreSQL:

• initial: لقطة عند أول تشغيل فقط، ثم التغييرات، وهي الافتراضية.
• no_data: دون بيانات قديمة، التغييرات من الآن فقط، وتناسب جدول Outbox.
• when_needed: لقطة عند أول تشغيل، ولقطة جديدة إذا اكتشف الموصل أن الموقع المحفوظ لم يعد موجوداً على السيرفر، وسوف نحتاجها في نهاية التجربة.

وللجداول الكبيرة هناك اللقطات التدريجية Incremental Snapshots، التي تقرأ الجدول على دفعات Chunks بالتوازي مع قراءة التغييرات، وتبدأ بإشارة تكتبها في جدول إشارات Signaling Table، فلا تحتاج أن توقف قراءة التغييرات ساعات حتى تنتهي لقطة جدول فيه مئات الملايين من الصفوف.

نمط Outbox مع Debezium

نقل جداول قاعدة البيانات كما هي إلى Kafka له عيب كبير: أنت تنشر شكل جداولك الداخلي، فكل الأنظمة التي تقرأ الموضوع shop.public.orders تعتمد على أسماء أعمدتك، وأي تغيير في الجدول يكسرها. والحل نمط Outbox الذي شرحناه في مشكلة الكتابة المزدوجة: الخدمة تكتب في المعاملة نفسها التغيير وحدثاً مصمماً للخارج في جدول outbox، وDebezium يقرأ هذا الجدول فقط.

والمحول Outbox Event Router يحول صف ال Outbox إلى رسالة نظيفة: الموضوع من العمود aggregatetype، والمفتاح من aggregateid فتبقى أحداث الطلب الواحد بالترتيب في Partition واحد، والقيمة هي payload وحده. ويذكر التوثيق أن جدول ال Outbox يعامل طابوراً، فالتعديل عليه غير مسموح، والحذف منه يتجاهله المحول، لذلك تستطيع أن تحذف الصفوف القديمة بمهمة دورية دون أن يظهر ذلك للمستهلكين.

تغيير شكل الجداول

يقول توثيق موصل PostgreSQL صراحة إن الفك المنطقي لا يدعم أوامر DDL، أي أن Debezium لا يرسل حدثاً عندما تضيف عموداً أو تغير اسمه، وإنما يظهر الشكل الجديد في أول حدث بعد التغيير. وهذا يعني:

• إضافة عمود: آمنة غالباً، فالمستهلك القديم يتجاهل الحقل الجديد.
• تغيير اسم عمود أو حذفه: يكسر كل مستهلك يقرأ الاسم القديم، دون أي خطأ عند Debezium، فالحقل يختفي من الأحداث فقط.
• تغيير النوع: قد يغير شكل القيمة في JSON، مثل رقم يصبح نصاً.

لذلك عامل أعمدة الجداول المنقولة كأنها API: أضف ولا تحذف ولا تغير الاسم مباشرة، وإذا احتجت تغييراً كبيراً فأضف العمود الجديد واكتب في الاثنين فترة ثم احذف القديم بعد أن تنتقل كل المستهلكين. وإذا كبر عدد المستهلكين فسجل الأشكال Schema Registry مع Avro أو Protobuf يرفض التغيير غير المتوافق قبل أن يصل إلى المستهلكين، أو استخدم نمط Outbox حتى لا ينشر شكل الجداول أصلاً.

الخطر الأكبر: ال Slot الذي ينتظر مستهلكاً متوقفاً

ال Replication Slot يمنع حذف ال WAL حتى يؤكد Debezium قراءته، ويقول توثيق الفك المنطقي إن ال Slot يحتفظ بال WAL حتى لو لم يكن أحد متصلاً به. فإذا توقف Kafka Connect يوماً بسبب عطل أو نسيان، فإن PostgreSQL يحتفظ بكل ال WAL لكل قاعدة البيانات، وليس فقط للجداول التي ينقلها Debezium، حتى يمتلئ القرص وتتوقف قاعدة البيانات الرئيسية نفسها. والصورة التالية تبين ذلك:

وهناك ثلاث أدوات للتعامل مع ذلك، ولكل منها ثمن:

• max_slot_wal_keep_size: الحد الأقصى لل WAL الذي يحتفظ به أي Slot، والقيمة الافتراضية -1 أي بلا حد بحسب توثيق إعدادات النسخ. إذا تجاوزه Slot أبطله PostgreSQL، فيحمي القرص لكن Debezium يفقد موقعه ويحتاج لقطة جديدة.
• idle_replication_slot_timeout: جديد في PostgreSQL 18، يبطل ال Slot الذي بقي دون اتصال أطول من مدة تحددها، والقيمة الافتراضية صفر أي معطل.
• نبضات Heartbeat: إذا كانت الجداول التي تنقلها هادئة بينما قاعدة البيانات مشغولة بجداول أخرى، فإن Debezium لا يستلم أي حدث ليؤكد موقعه، فيكبر ال WAL رغم أن Debezium يعمل. ويشرح التوثيق أن الحل أن تكتب في جدول صغير بانتظام عبر الإعداد heartbeat.action.query حتى يتقدم الموقع.

والقاعدة العملية: راقب حجم ال WAL المحفوظ لكل Slot بالاستعلام slots.sql في التجربة، ونبه عندما يتجاوز حداً صغيراً، وضع max_slot_wal_keep_size أكبر من أطول توقف تتوقعه وأصغر من مساحة القرص الحرة، واحذف أي Slot لم يعد له مستهلك.

تجربة عملية: من اللقطة الأولى إلى ال Slot الضائع

سوف نشغل PostgreSQL 18.6 وApache Kafka 4.3.1 بنمط KRaft بعقدة واحدة، وDebezium 3.7.0.Final داخل Kafka Connect 4.3.1 من Image الرسمية quay.io/debezium/connect، على Docker Engine 29.8.0. ولاحظ أن آخر Image لDebezium على Docker Hub هي 3.0.0.Final من أكتوبر 2024، والإصدارات بعدها على quay.io فقط.

ملفات التجربة

الملف compose.yaml:

name: lab-q2-dbz

services:
  postgres:
    image: postgres:18.6
    container_name: lab-q2-dbz-pg
    environment:
      POSTGRES_USER: shop
      POSTGRES_PASSWORD: shop-lab-pass
      POSTGRES_DB: shop
    command:
      - postgres
      - -c
      - wal_level=logical            # logical decoding needs this; the default is replica
      - -c
      - max_wal_size=256MB           # lab only: frequent checkpoints so WAL growth is easy to see
    volumes:
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql:ro
      - pg:/var/lib/postgresql
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "shop"]
      interval: 3s
      retries: 30

  kafka:
    image: apache/kafka:4.3.1
    container_name: lab-q2-dbz-kafka
    hostname: kafka
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_HEAP_OPTS: "-Xms512m -Xmx512m"
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 > /dev/null 2>&1"]
      interval: 5s
      retries: 40

  connect:
    image: quay.io/debezium/connect:3.7.0.Final
    container_name: lab-q2-dbz-connect
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: lab-connect
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_status
      HEAP_OPTS: "-Xms512m -Xmx512m"
    ports: ["127.0.0.1:18083:8083"]              # Kafka Connect REST API, from this machine only
    depends_on:
      kafka: { condition: service_healthy }
      postgres: { condition: service_healthy }

volumes:
  pg:

والملف init.sql يبني قاعدة متجر صغيرة، وفيها جدول page_views لا ينقله أحد، سوف نستخدمه لنملأ ال WAL:

-- Shop database for the CDC lab (ArabRoot)
CREATE TABLE customers (
  id    serial PRIMARY KEY,
  name  text NOT NULL,
  email text NOT NULL UNIQUE,
  city  text
);
CREATE TABLE orders (
  id          serial PRIMARY KEY,
  customer_id int NOT NULL REFERENCES customers(id),
  total       numeric(10,2) NOT NULL,
  status      text NOT NULL DEFAULT 'new'
);
-- Outbox: the service writes its events here in the same transaction as the business change
CREATE TABLE outbox (
  id            uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  aggregatetype text NOT NULL,         -- decides the topic: outbox.event.<aggregatetype>
  aggregateid   text NOT NULL,         -- becomes the Kafka message key, so one order keeps its order
  type          text NOT NULL,         -- event name, sent as a header
  payload       jsonb NOT NULL
);
-- A busy table that nobody captures, used to show that a slot keeps all WAL, not only captured tables
CREATE TABLE page_views (id bigserial PRIMARY KEY, url text, at timestamptz DEFAULT now(), filler text);

INSERT INTO customers (name, email, city) VALUES
  ('Sara', '[email protected]', 'Riyadh'),
  ('Omar', '[email protected]', 'Jeddah'),
  ('Lina', '[email protected]', 'Dammam');
INSERT INTO orders (customer_id, total) VALUES (1, 250.00), (2, 99.50);

والموصل الأول shop-connector.json ينقل جدولي العملاء والطلبات، مع JSON مقروء دون الأشكال المضمنة:

{
  "name": "shop-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "shop",
    "database.password": "shop-lab-pass",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "plugin.name": "pgoutput",
    "slot.name": "shop_cdc",
    "publication.name": "shop_cdc_pub",
    "publication.autocreate.mode": "filtered",
    "table.include.list": "public.customers,public.orders",
    "snapshot.mode": "initial",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "decimal.handling.mode": "string"
  }
}

والموصل الثاني outbox-connector.json يقرأ جدول ال Outbox فقط، بال Slot الخاص به والمحول Outbox Event Router:

{
  "name": "shop-outbox",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "shop",
    "database.password": "shop-lab-pass",
    "database.dbname": "shop",
    "topic.prefix": "outboxsrc",
    "plugin.name": "pgoutput",
    "slot.name": "shop_outbox",
    "publication.name": "shop_outbox_pub",
    "publication.autocreate.mode": "filtered",
    "table.include.list": "public.outbox",
    "snapshot.mode": "no_data",
    "tombstones.on.delete": "false",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.expand.json.payload": "true",
    "transforms.outbox.table.fields.additional.placement": "type:header:eventType",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

وثلاثة ملفات صغيرة للمساعدة:

# Prints Debezium change events from kafka-console-consumer output in a short form (ArabRoot lab).
#   ... kafka-console-consumer.sh ... --formatter-property print.key=true | python show.py
import json, sys
for line in sys.stdin:
    if '\t' not in line: continue
    key, value = line.rstrip('\n').split('\t', 1)
    if value == 'null':
        print(f'key={key} value=null (tombstone)'); continue
    e = json.loads(value)
    if 'op' in e:
        print(f'op={e["op"]} key={key} before={json.dumps(e["before"], ensure_ascii=False)} after={json.dumps(e["after"], ensure_ascii=False)} lsn={e["source"]["lsn"]}')
    else:
        print(f'key={key} value={json.dumps(e, ensure_ascii=False)}')

-- How much WAL each replication slot keeps on disk (ArabRoot lab)
SELECT slot_name, active, wal_status,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal,
       pg_size_pretty(safe_wal_size) AS safe_wal_size
FROM pg_replication_slots ORDER BY slot_name;
SELECT pg_size_pretty(sum(size)) AS pg_wal_dir FROM pg_ls_waldir();

-- About 250 MB of WAL from a table that no connector captures (ArabRoot lab)
INSERT INTO page_views (url, filler)
SELECT '/product/' || (g % 500), repeat(md5(g::text), 16) FROM generate_series(1, 300000) g;
CHECKPOINT;

التشغيل واللقطة الأولى

docker compose up -d --wait
curl -s -X POST -H "Content-Type: application/json" --data @shop-connector.json http://127.0.0.1:18083/connectors
curl -s http://127.0.0.1:18083/connectors/shop-cdc/status
docker exec lab-q2-dbz-pg psql -U shop -c "select slot_name, plugin, active, confirmed_flush_lsn from pg_replication_slots"

والمخرج سوف يكون كما يلي:

{"name":"shop-cdc","connector":{"state":"RUNNING","worker_id":"192.168.48.4:8083","version":"3.7.0.Final"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"192.168.48.4:8083","version":"3.7.0.Final"}],"type":"source"}
 slot_name |  plugin  | active | confirmed_flush_lsn
-----------+----------+--------+---------------------
 shop_cdc  | pgoutput | t      | 0/1C25138

وأول حدث في الموضوع shop.public.customers كما هو دون اختصار:

{"id":1}	{"before":null,"after":{"id":1,"name":"Sara","email":"[email protected]","city":"Riyadh"},"source":{"version":"3.7.0.Final","connector":"postgresql","name":"shop","ts_ms":1791818892979,"snapshot":"first","db":"shop","sequence":"[null,\"29511992\"]","ts_us":1791818892979280,"ts_ns":1791818892979280000,"schema":"public","table":"customers","txId":772,"lsn":29511992,"xmin":null,"origin":null,"origin_lsn":null},"transaction":null,"op":"r","ts_ms":1791818893175,"ts_us":1791818893175830,"ts_ns":1791818893175830024}

لاحظ أن op قيمته r لأن الصف جاء من اللقطة، وأن snapshot قيمته first لأول صف فيها، وأن Debezium أنشأ ال Publication وال Slot بنفسه.

الإضافة والتعديل والحذف

P="docker exec lab-q2-dbz-pg psql -U shop -q -c"
$P "INSERT INTO customers (name, email, city) VALUES ('Huda', '[email protected]', 'Abha')"
$P "UPDATE customers SET city = 'Khobar' WHERE id = 2"
$P "UPDATE orders SET status = 'paid' WHERE id = 1"
$P "DELETE FROM orders WHERE id = 2"
$P "ALTER TABLE customers REPLICA IDENTITY FULL"
$P "UPDATE customers SET city = 'Taif' WHERE id = 3"
for t in customers orders; do
  docker exec lab-q2-dbz-kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 \
    --topic shop.public.$t --from-beginning --formatter-property print.key=true --timeout-ms 6000 | python show.py
done

والمخرج سوف يكون كما يلي:

op=r key={"id":1} before=null after={"id": 1, "name": "Sara", "email": "[email protected]", "city": "Riyadh"} lsn=29511992
op=r key={"id":2} before=null after={"id": 2, "name": "Omar", "email": "[email protected]", "city": "Jeddah"} lsn=29511992
op=r key={"id":3} before=null after={"id": 3, "name": "Lina", "email": "[email protected]", "city": "Dammam"} lsn=29511992
op=c key={"id":4} before=null after={"id": 4, "name": "Huda", "email": "[email protected]", "city": "Abha"} lsn=29532376
op=u key={"id":2} before=null after={"id": 2, "name": "Omar", "email": "[email protected]", "city": "Khobar"} lsn=29532984
op=u key={"id":3} before={"id": 3, "name": "Lina", "email": "[email protected]", "city": "Dammam"} after={"id": 3, "name": "Lina", "email": "[email protected]", "city": "Taif"} lsn=29533824
op=r key={"id":1} before=null after={"id": 1, "customer_id": 1, "total": "250.00", "status": "new"} lsn=29511992
op=r key={"id":2} before=null after={"id": 2, "customer_id": 2, "total": "99.50", "status": "new"} lsn=29511992
op=u key={"id":1} before=null after={"id": 1, "customer_id": 1, "total": "250.00", "status": "paid"} lsn=29533136
op=d key={"id":2} before={"id": 2, "customer_id": 0, "total": "0.00", "status": "new"} after=null lsn=29533272
key={"id":2} value=null (tombstone)

في المخرج أعلاه لاحظ التالي:

• تعديل مدينة Omar وصل وbefore فارغ، لأن الجدول كان بالإعداد الافتراضي، وبعد REPLICA IDENTITY FULL وصل تعديل Lina ومعه المدينة القديمة Dammam.
• حذف الطلب 2 هو الأخطر: before فيه المفتاح الصحيح فقط، أما customer_id فصفر وtotal صفر وstatus قيمته new، وهي القيمة الافتراضية للعمود وليست قيمة الصف. فإذا كان مستهلك التقارير يطرح total من مجموع العميل عند الحذف فسوف يطرح صفراً من العميل صفر دون أي خطأ.
• بعد حدث الحذف جاء حدث قيمته null يسمى Tombstone، وهو لمواضيع Kafka المضغوطة Compacted حتى يحذف المفتاح نهائياً منها.

ال Outbox: معاملتان تنشران، وواحدة تتراجع

curl -s -X POST -H "Content-Type: application/json" --data @outbox-connector.json http://127.0.0.1:18083/connectors
docker exec -i lab-q2-dbz-pg psql -U shop -q <<'EOF'
BEGIN;
INSERT INTO orders (customer_id, total) VALUES (4, 410.00);
INSERT INTO outbox (aggregatetype, aggregateid, type, payload)
VALUES ('order', currval('orders_id_seq')::text, 'OrderPlaced',
        jsonb_build_object('order_id', currval('orders_id_seq'), 'customer_id', 4, 'total', 410.00));
COMMIT;
BEGIN;
UPDATE orders SET status = 'paid' WHERE id = 3;
INSERT INTO outbox (aggregatetype, aggregateid, type, payload)
VALUES ('order', '3', 'OrderPaid', '{"order_id": 3, "method": "mada"}');
COMMIT;
BEGIN;
INSERT INTO orders (customer_id, total) VALUES (1, 75.00);
INSERT INTO outbox (aggregatetype, aggregateid, type, payload)
VALUES ('order', currval('orders_id_seq')::text, 'OrderPlaced', '{"order_id": 4, "customer_id": 1, "total": 75.00}');
ROLLBACK;
EOF
docker exec lab-q2-dbz-kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic outbox.event.order \
  --from-beginning --formatter-property print.key=true --formatter-property print.headers=true --timeout-ms 6000

والمخرج بعد اختصار ترويسات Debezium الداخلية سوف يكون كما يلي:

...,id:6562fe7f-8a2f-4376-8bc3-7fbdd57494ae,eventType:OrderPlaced	3	{"total":410.0,"order_id":3,"customer_id":4}
...,id:7e2a8144-5f5d-4653-9910-4a372d55a4c6,eventType:OrderPaid	3	{"method":"mada","order_id":3}

في المخرج أعلاه لاحظ التالي:

• الموضوع outbox.event.order جاء من قيمة aggregatetype، والمفتاح 3 من aggregateid، فالحدثان للطلب نفسه في Partition واحد وبالترتيب.
• القيمة هي ال payload فقط، ونوع الحدث في ترويسة eventType ومعرفه في id، فالمستهلك يستخدم المعرف ليتجاهل التكرار.
• المعاملة الثالثة تراجعت بالأمر ROLLBACK، فلم يظهر لها حدث، لأن الفك المنطقي يرسل المعاملات الملتزمة فقط، وهذا هو الضمان الذي يحل مشكلة الكتابة المزدوجة.

تغيير شكل الجدول

$P "ALTER TABLE customers ADD COLUMN phone text"
$P "UPDATE customers SET phone = '0500000001' WHERE id = 1"
$P "ALTER TABLE customers RENAME COLUMN city TO region"
$P "UPDATE customers SET region = 'Makkah' WHERE id = 4"

وآخر حدثين في موضوع العملاء:

op=u key={"id":1} before={"id": 1, "name": "Sara", "email": "[email protected]", "city": "Riyadh", "phone": null} after={"id": 1, "name": "Sara", "email": "[email protected]", "city": "Riyadh", "phone": "0500000001"} lsn=29552960
op=u key={"id":4} before={"id": 4, "name": "Huda", "email": "[email protected]", "region": "Abha", "phone": null} after={"id": 4, "name": "Huda", "email": "[email protected]", "region": "Makkah", "phone": null} lsn=29561928

الحقل phone ظهر في أول حدث بعد إضافته، ولم يصل أي حدث للأمر ALTER TABLE نفسه. وبعد تغيير الاسم اختفى city من الأحداث وظهر region، وظل الموصل في حالة RUNNING، فكل مستهلك يقرأ after["city"] سوف يجد قيمة فارغة أو يتوقف بخطأ، وDebezium لم يحذرك.

نكسرها: Kafka Connect يتوقف والقرص يمتلئ

نسجل حالة ال Slots، ثم نوقف Kafka Connect، ونكتب في جدول page_views الذي لا ينقله أي موصل 900 ألف صف على ثلاث دفعات:

docker cp slots.sql lab-q2-dbz-pg:/tmp/slots.sql; docker cp noise.sql lab-q2-dbz-pg:/tmp/noise.sql
docker exec lab-q2-dbz-pg psql -U shop -f /tmp/slots.sql
docker stop lab-q2-dbz-connect
for i in 1 2 3; do docker exec lab-q2-dbz-pg psql -U shop -q -f /tmp/noise.sql; done
docker exec lab-q2-dbz-pg psql -U shop -f /tmp/slots.sql

والمخرج سوف يكون كما يلي:

  slot_name  | active | wal_status | retained_wal | safe_wal_size
-------------+--------+------------+--------------+---------------
 shop_cdc    | t      | reserved   | 57 kB        |
 shop_outbox | t      | reserved   | 53 kB        |

 pg_wal_dir
------------
 16 MB

  slot_name  | active | wal_status | retained_wal | safe_wal_size
-------------+--------+------------+--------------+---------------
 shop_cdc    | f      | extended   | 583 MB       |
 shop_outbox | f      | extended   | 583 MB       |

 pg_wal_dir
------------
 608 MB

في المخرج أعلاه لاحظ التالي:

• كبر مجلد ال WAL من 16 إلى 608 ميجابايت، رغم أن الإعداد max_wal_size هو 256 ميجابايت، لأن ال Slots منعت حذفه.
• كل الكتابة كانت في جدول لا ينقله أي موصل، ومع ذلك احتفظ PostgreSQL بها كلها، لأن ال WAL واحد لكل قاعدة البيانات.
• الحالة extended تعني أن ال WAL المحفوظ تجاوز max_wal_size، ولا يوجد أي حد يوقفه، فلو بقي Kafka Connect متوقفاً أسبوعاً على قاعدة بيانات تكتب جيجابايتات يومياً لامتلأ القرص وتوقفت قاعدة البيانات الرئيسية.

نضع حداً، فيضيع ال Slot

الآن نضع حداً أقصى 128 ميجابايت، ونكتب دفعة أخرى، ثم نعدل عميلاً بينما Kafka Connect ما زال متوقفاً، ثم نشغله:

docker exec lab-q2-dbz-pg psql -U shop -c "ALTER SYSTEM SET max_slot_wal_keep_size = '128MB'" -c "SELECT pg_reload_conf()"
docker exec lab-q2-dbz-pg psql -U shop -f /tmp/slots.sql
docker exec lab-q2-dbz-pg psql -U shop -q -f /tmp/noise.sql
docker exec lab-q2-dbz-pg psql -U shop -f /tmp/slots.sql
docker logs lab-q2-dbz-pg 2>&1 | grep -A2 'invalidating obsolete replication slot "shop_cdc"'
docker exec lab-q2-dbz-pg psql -U shop -q -c "UPDATE customers SET region = 'Tabuk' WHERE id = 1"
docker start lab-q2-dbz-connect

والمخرج بعد حذف الوقت من بداية أسطر السجل سوف يكون كما يلي:

  slot_name  | active | wal_status | retained_wal | safe_wal_size
-------------+--------+------------+--------------+---------------
 shop_cdc    | f      | unreserved | 583 MB       | -451 MB
 shop_outbox | f      | unreserved | 583 MB       | -451 MB

  slot_name  | active | wal_status | retained_wal | safe_wal_size
-------------+--------+------------+--------------+---------------
 shop_cdc    | f      | lost       |              |
 shop_outbox | f      | lost       |              |

 pg_wal_dir
------------
 256 MB

LOG:  invalidating obsolete replication slot "shop_cdc"
DETAIL:  The slot's restart_lsn 0/1C2F108 exceeds the limit by 675090168 bytes.
HINT:  You might need to increase "max_slot_wal_keep_size".

بعد الحد أصبحت الحالة unreserved، أي أن ال WAL المحفوظ أكبر من الحد وسوف يحذف في نقطة الحفظ Checkpoint التالية، ثم جاءت نقطة الحفظ فأبطل PostgreSQL ال Slots وأصبحت الحالة lost، وعاد مجلد ال WAL إلى 256 ميجابايت. فالقرص نجا، لكن ماذا حدث لDebezium؟ بعد تشغيل Kafka Connect ظل سجله يكتب هذا السطر كل ثانيتين، وحالة المهمة في ال REST API كانت UNASSIGNED وليست FAILED:

WARN   Postgres||lifecycle  Cannot obtain valid replication slot 'shop_cdc' for plugin 'pgoutput' and database 'shop' [during attempt 115 out of 900, concurrent tx probably blocks taking snapshot.

لاحظ أن الرسالة مضللة، فهي تقترح أن معاملة أخرى تمنع اللقطة، بينما السبب الحقيقي أن ال Slot أبطل. ولو كانت المراقبة على حالة FAILED فقط لما ظهر أي تنبيه. وبعد أن حذفنا ال Slot الضائع بالأمر pg_drop_replication_slot تحولت المهمة إلى FAILED برسالة واضحة هذه المرة، بعد اختصار تفاصيل الموقع:

io.debezium.DebeziumException: The connector is trying to read change stream starting at PostgresOffsetContext [...], but this is no longer available on the server. Reconfigure the connector to use a snapshot mode when needed.

الاستعادة بلقطة جديدة، وما الذي ضاع فعلاً

نغير snapshot.mode إلى when_needed ونعيد تشغيل المهمة:

python -c "import json; c = json.load(open('shop-connector.json'))['config']; c['snapshot.mode'] = 'when_needed'; print(json.dumps(c))" > cfg.json
curl -s -X PUT -H "Content-Type: application/json" --data @cfg.json http://127.0.0.1:18083/connectors/shop-cdc/config
curl -s -X POST "http://127.0.0.1:18083/connectors/shop-cdc/restart?includeTasks=true&onlyFailed=true"

وآخر أحداث موضوع العملاء وحالة ال Slots سوف تكون كما يلي:

op=r key={"id":2} before=null after={"id": 2, "name": "Omar", "email": "[email protected]", "region": "Khobar", "phone": null} lsn=1563853864
op=r key={"id":3} before=null after={"id": 3, "name": "Lina", "email": "[email protected]", "region": "Taif", "phone": null} lsn=1563853864
op=r key={"id":4} before=null after={"id": 4, "name": "Huda", "email": "[email protected]", "region": "Makkah", "phone": null} lsn=1563853864
op=r key={"id":1} before=null after={"id": 1, "name": "Sara", "email": "[email protected]", "region": "Tabuk", "phone": "0500000001"} lsn=1563853864
  slot_name  | active | wal_status | retained_wal | safe_wal_size
-------------+--------+------------+--------------+---------------
 shop_cdc    | t      | reserved   | 1328 bytes   | 141 MB
 shop_outbox | f      | lost       |              |

في المخرج أعلاه لاحظ التالي:

• الموصل أنشأ Slot جديداً، وأرسل كل العملاء من جديد بأحداث op=r، ومنها التعديل الذي حدث أثناء التوقف (Tabuk)، فالحالة الأخيرة وصلت.
• لكن الذي وصل هو الحالة الأخيرة فقط، فلو تغير الصف مرتين أثناء التوقف لفات المستهلك التغيير الأول، ولو حذف صف أثناء التوقف فلن يصل حدث حذفه أبداً، لأن اللقطة تقرأ الصفوف الموجودة فقط، فيبقى الصف المحذوف في محرك البحث أو ال Cache. لذلك بعد أي لقطة استعادة يحتاج كل مستهلك أن يقارن بياناته أو يعيد بناءها.
• ال Slot الخاص بال Outbox ما زال lost، وهنا الخسارة أكبر، لأن أحداث ال Outbox التي كتبت أثناء التوقف لن تعود باللقطة إذا كانت صفوفها حذفت، وهذا سبب إضافي لتحذف صفوف ال Outbox القديمة بعد مدة أطول من أطول توقف تتوقعه، لا بعد ثوان.

حذف المختبر

docker compose down -v
docker image rm quay.io/debezium/connect:3.7.0.Final

قبل الإنتاج

• مستخدم بصلاحيات محدودة: لا تستخدم مالك قاعدة البيانات كما في التجربة، وأنشئ مستخدماً بصلاحية REPLICATION وقراءة الجداول المنقولة فقط، وأنشئ ال Publication بنفسك بدل أن تعطيه صلاحية إنشائها.
• المراقبة: حجم ال WAL المحفوظ لكل Slot، وحالة wal_status، وحالة مهام Kafka Connect بما فيها UNASSIGNED، والتأخر بين وقت الحدث في source.ts_ms ووقت وصوله.
• الحد الأقصى وال Heartbeat: ضع max_slot_wal_keep_size بقيمة تتحمل أطول توقف متوقع، وفعل heartbeat.action.query إذا كانت الجداول المنقولة هادئة.
• Failover في PostgreSQL: ال Slot يعيش على السيرفر الرئيسي، فعند انتقال الرئاسة إلى نسخة أخرى تحتاج أن يكون ال Slot موجوداً عليها، فاختبر هذا السيناريو مع أداة التوفر العالي التي تستخدمها قبل أن تعتمد عليه.
• المستهلكون آمنون عند التكرار: بعد أي إعادة تشغيل قد يرسل Debezium بعض الأحداث مرة ثانية، وبعد لقطة الاستعادة يرسل كل الصفوف، فاجعل المستهلك يكتب بالمفتاح Upsert وليس إضافة عمياء.

أخطاء شائعة

• Slot بلا مستهلك: تجربة قديمة أو موصل محذوف ترك Slot خلفه، فيكبر ال WAL حتى يمتلئ القرص، وأول ما تفعله عند أي مشكلة قرص في PostgreSQL أن تنظر في pg_replication_slots.
• الاعتماد على before دون REPLICA IDENTITY FULL: التعديل يصل دونه، والحذف يصل بقيم افتراضية تبدو حقيقية.
• تغيير اسم عمود منقول: Debezium يستمر دون خطأ، والمستهلكون ينكسرون بصمت.
• مراقبة حالة FAILED فقط: رأيت أن ال Slot الضائع أبقى المهمة في حالة UNASSIGNED برسالة مضللة.
• نشر جداولك الداخلية لكل الفرق: شكل الجدول يصبح عقداً لا تستطيع تغييره، واستخدم Outbox للأحداث التي تقرؤها فرق أخرى.
• Images قديمة من Docker Hub: توقف النشر هناك عند 3.0.0.Final، والإصدارات الحالية على quay.io.

الخلاصة

وصلنا لنهاية الموضوع، وأهم ما فيه:

• CDC بقراءة السجل يرى كل إضافة وتعديل وحذف بترتيب الالتزام دون أي تعديل في التطبيق، وDebezium يفعل ذلك مع PostgreSQL عبر الفك المنطقي وال Replication Slot.
• اللقطة الأولى تنقل البيانات الموجودة، ثم يكمل من التغييرات دون فجوة، والحذف والتعديل يحتاجان REPLICA IDENTITY FULL إذا أردت القيم القديمة.
• مع نمط Outbox وال Event Router تنشر أحداثاً مصممة للخارج بالمفتاح الصحيح، والمعاملة التي تتراجع لا تنشر شيئاً.
• DDL لا يرسل حدثاً، وتغيير اسم عمود يكسر المستهلكين بصمت.
• في تجربتنا كبر ال WAL من 16 إلى 608 ميجابايت بسبب كتابة في جدول غير منقول بينما Kafka Connect متوقف، ومع حد 128 ميجابايت ضاع ال Slot، وأعطى Debezium رسالة مضللة، والاستعادة بلقطة جديدة أعادت الحالة الأخيرة فقط.

دليلالمعاملات الموزعة: Saga ونمط Outbox بدل 2PCكيف تنفذ عملية تمتد على عدة خدمات دون 2PC، بخطوات محلية وتعويضات وجدول Outbox لا يضيع فيه أي حدث.دليلApache Kafka: بث الأحداث بأحجام ضخمةكيف يحفظ عنقود Kafka أحداثك بنسخ على عدة سيرفرات، وماذا يحدث للمنتجين والمستهلكين عندما يسقط أحدها.دليلAudit Trail: كيف تبني سجل تدقيق لا يقبل العبثكيف تصمم سجلاً يجيب عن سؤال من غير هذه الصلاحية ومتى، وتحميه من الحذف والتعديل بعد كتابته.دليلRabbitMQ أم Kafka أم NATS؟ كيف تختار وسيط الرسائلكيف تختار بين RabbitMQ وKafka وNATS بحسب نموذج كل منها وضماناته وتكلفة تشغيله، مع قياس على جهاز واحد.

سجل التحديثات

• أكتوبر 2026: كتابة المقال، وتنفيذ التجربة على PostgreSQL 18.6 وApache Kafka 4.3.1 وDebezium 3.7.0.Final داخل Kafka Connect 4.3.1 في Docker.