> ## Content Index
> Fetch the complete content index at: https://arabroot.io/llms.txt
> Use this file to discover other available public pages before exploring further.

# نقل تغييرات قاعدة البيانات CDC مع Debezium
- URL: https://arabroot.io/articles/نقل-تغييرات-قاعدة-البيانات-cdc-مع-debezium/
- Published: 2026-10-12T23:15:00.000Z
- Updated: 2026-10-12T16:08:56.000Z
- Description: كيف تنقل كل تغيير في PostgreSQL إلى Kafka مع Debezium ونمط Outbox، ولماذا قد يملأ مستهلك متوقف قرص قاعدة بياناتك.
- Author: فريق عرب رووت
- Tags: الأدلة التقنية, هندسة البرمجيات وتصميم الأنظمة, قواعد البيانات, DevOps وCI/CD, الاستضافة الذاتية

في مقال [Saga ونمط Outbox](https://arabroot.io/articles/%D8%A7%D9%84%D9%85%D8%B9%D8%A7%D9%85%D9%84%D8%A7%D8%AA-%D8%A7%D9%84%D9%85%D9%88%D8%B2%D8%B9%D8%A9-saga-%D9%88%D9%86%D9%85%D8%B7-outbox/#relay) ذكرنا طريقتين لإخراج الأحداث من جدول ال 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:** كل تغيير يكتب صفاً في جدول آخر، وقد شرحناها في مقال [سجل التدقيق](https://arabroot.io/articles/%D8%B3%D8%AC%D9%84-%D8%A7%D9%84%D8%AA%D8%AF%D9%82%D9%8A%D9%82-audit-trail-%D9%85%D9%86-%D9%81%D8%B9%D9%84-%D9%85%D8%A7%D8%B0%D8%A7-%D9%88%D9%85%D8%AA%D9%89/). ترى كل شيء، لكنها تضيف كتابة على كل معاملة، وما زلت تحتاج من ينقل الجدول إلى الأنظمة الأخرى.
- **قراءة السجل Log-Based CDC:** قاعدة البيانات تكتب كل تغيير في سجل المعاملات قبل تنفيذه، وهو ال WAL في PostgreSQL، والأداة تقرأ هذا السجل. ترى كل إضافة وتعديل وحذف بترتيب الالتزام Commit، ولا تضيف حملاً على الجداول نفسها، ولا تحتاج أي تعديل في كود التطبيق. وهذه طريقة Debezium.

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

Debezium مشروع مفتوح المصدر برخصة Apache 2.0 بحسب [مستودعه](https://github.com/debezium/debezium?ref=arabroot.io)، وآخر إصدار عند كتابة المقال 3.7.0.Final الذي صدر في 29 سبتمبر 2026، وتذكر [صفحة الإصدار](https://debezium.io/releases/3.7/?ref=arabroot.io) أنه مختبر مع PostgreSQL من 14 إلى 18 ومع Kafka Connect 3.1 وما بعده. والصورة التالية تبين الأجزاء:

![التطبيق يكتب في PostgreSQL، وكل تغيير يكتب في WAL. ال publication تحدد الجداول customers وorders، وال replication slot اسمه shop_cdc ويحفظ موقع القراءة confirmed_flush_lsn. يقرأ Debezium داخل Kafka Connect التغييرات عبر pgoutput، ويكتب الأحداث إلى المواضيع shop.public.customers وshop.public.orders، ويحفظ ال offset في الموضوع connect_offsets، ثم يقرأ البحث والتقارير وال cache من Kafka](https://arabroot.io/content/images/2026/10/debezium-cdc-01-pipeline.webp)

Debezium يقرأ ال WAL عبر Replication Slot، ويحول كل صف متغير إلى حدث في موضوع خاص بجدوله

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](https://debezium.io/documentation/reference/stable/operations/debezium-server.html?ref=arabroot.io)، وهو برنامج مستقل يقرأ التغييرات ويرسلها مباشرة إلى وجهات كثيرة منها 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](https://debezium.io/documentation/reference/stable/connectors/postgresql.html?ref=arabroot.io):

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

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

## نمط Outbox مع Debezium

نقل جداول قاعدة البيانات كما هي إلى Kafka له عيب كبير: أنت تنشر شكل جداولك الداخلي، فكل الأنظمة التي تقرأ الموضوع `shop.public.orders` تعتمد على أسماء أعمدتك، وأي تغيير في الجدول يكسرها. والحل نمط Outbox الذي شرحناه في [مشكلة الكتابة المزدوجة](https://arabroot.io/articles/%D8%A7%D9%84%D9%85%D8%B9%D8%A7%D9%85%D9%84%D8%A7%D8%AA-%D8%A7%D9%84%D9%85%D9%88%D8%B2%D8%B9%D8%A9-saga-%D9%88%D9%86%D9%85%D8%B7-outbox/#dual-write): الخدمة تكتب في المعاملة نفسها التغيير وحدثاً مصمماً للخارج في جدول `outbox`، وDebezium يقرأ هذا الجدول فقط.

والمحول [Outbox Event Router](https://debezium.io/documentation/reference/stable/transformations/outbox-event-router.html?ref=arabroot.io) يحول صف ال Outbox إلى رسالة نظيفة: الموضوع من العمود `aggregatetype`، والمفتاح من `aggregateid` فتبقى أحداث الطلب الواحد بالترتيب في Partition واحد، والقيمة هي `payload` وحده. ويذكر التوثيق أن جدول ال Outbox يعامل طابوراً، فالتعديل عليه غير مسموح، والحذف منه يتجاهله المحول، لذلك تستطيع أن تحذف الصفوف القديمة بمهمة دورية دون أن يظهر ذلك للمستهلكين.

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

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

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

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

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

ال Replication Slot يمنع حذف ال WAL حتى يؤكد Debezium قراءته، ويقول [توثيق الفك المنطقي](https://www.postgresql.org/docs/18/logicaldecoding-explanation.html?ref=arabroot.io) إن ال Slot يحتفظ بال WAL حتى لو لم يكن أحد متصلاً به. فإذا توقف Kafka Connect يوماً بسبب عطل أو نسيان، فإن PostgreSQL يحتفظ بكل ال WAL لكل قاعدة البيانات، وليس فقط للجداول التي ينقلها Debezium، حتى يمتلئ القرص وتتوقف قاعدة البيانات الرئيسية نفسها. والصورة التالية تبين ذلك:

![ثلاث مراحل من التجربة. الأولى Debezium يعمل وال slot نشط وwal_status reserved وال WAL المحفوظ 57 kB ومجلد pg_wal حجمه 16 MB. الثانية Kafka Connect متوقف وكتبنا 3 مرات 300 ألف صف في جدول page_views الذي لا ينقله أحد، فأصبح wal_status extended وال WAL المحفوظ 583 MB ومجلد pg_wal 608 MB. الثالثة بعد max_slot_wal_keep_size بقيمة 128MB وكتابة إضافية، أصبح wal_status lost وعاد pg_wal إلى 256 MB، لكن الموصل لا يستطيع أن يكمل ويحتاج snapshot جديداً](https://arabroot.io/content/images/2026/10/debezium-cdc-02-slot-growth.webp)

ال Slot يحمي التغييرات من الضياع، وثمنه أن القرص يمتلئ إذا توقف المستهلك، والحد الأقصى يحمي القرص ويضيع ال Slot

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

- `max_slot_wal_keep_size`: الحد الأقصى لل WAL الذي يحتفظ به أي Slot، والقيمة الافتراضية `-1` أي بلا حد بحسب [توثيق إعدادات النسخ](https://www.postgresql.org/docs/18/runtime-config-replication.html?ref=arabroot.io). إذا تجاوزه 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](https://hub.docker.com/r/debezium/connect/tags?ref=arabroot.io) هي 3.0.0.Final من أكتوبر 2024، والإصدارات بعدها على quay.io فقط.

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

الملف `compose.yaml`:

```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:

```sql
-- 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', 'sara@example.com', 'Riyadh'),
  ('Omar', 'omar@example.com', 'Jeddah'),
  ('Lina', 'lina@example.com', 'Dammam');
INSERT INTO orders (customer_id, total) VALUES (1, 250.00), (2, 99.50);
```

والموصل الأول `shop-connector.json` ينقل جدولي العملاء والطلبات، مع 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:

```json
{
  "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"
  }
}
```

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

`show.py`: يعرض الأحداث في سطر واحد لكل حدث 

```python
# 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)}')
```

`slots.sql` و`noise.sql`: حجم ال WAL لكل Slot، وكتابة كثيرة في جدول غير منقول 

```sql
-- 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;
```

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

```bash
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"
```

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

```text
{"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` كما هو دون اختصار:

```json
{"id":1}	{"before":null,"after":{"id":1,"name":"Sara","email":"sara@example.com","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 بنفسه.

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

```bash
P="docker exec lab-q2-dbz-pg psql -U shop -q -c"
$P "INSERT INTO customers (name, email, city) VALUES ('Huda', 'huda@example.com', '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
```

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

```text
op=r key={"id":1} before=null after={"id": 1, "name": "Sara", "email": "sara@example.com", "city": "Riyadh"} lsn=29511992
op=r key={"id":2} before=null after={"id": 2, "name": "Omar", "email": "omar@example.com", "city": "Jeddah"} lsn=29511992
op=r key={"id":3} before=null after={"id": 3, "name": "Lina", "email": "lina@example.com", "city": "Dammam"} lsn=29511992
op=c key={"id":4} before=null after={"id": 4, "name": "Huda", "email": "huda@example.com", "city": "Abha"} lsn=29532376
op=u key={"id":2} before=null after={"id": 2, "name": "Omar", "email": "omar@example.com", "city": "Khobar"} lsn=29532984
op=u key={"id":3} before={"id": 3, "name": "Lina", "email": "lina@example.com", "city": "Dammam"} after={"id": 3, "name": "Lina", "email": "lina@example.com", "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: معاملتان تنشران، وواحدة تتراجع

```bash
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 الداخلية سوف يكون كما يلي:

```text
...,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`، فلم يظهر لها حدث، لأن الفك المنطقي يرسل المعاملات الملتزمة فقط، وهذا هو الضمان الذي يحل مشكلة الكتابة المزدوجة.

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

```bash
$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"
```

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

```text
op=u key={"id":1} before={"id": 1, "name": "Sara", "email": "sara@example.com", "city": "Riyadh", "phone": null} after={"id": 1, "name": "Sara", "email": "sara@example.com", "city": "Riyadh", "phone": "0500000001"} lsn=29552960
op=u key={"id":4} before={"id": 4, "name": "Huda", "email": "huda@example.com", "region": "Abha", "phone": null} after={"id": 4, "name": "Huda", "email": "huda@example.com", "region": "Makkah", "phone": null} lsn=29561928
```

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

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

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

```bash
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
```

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

```text
  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 ما زال متوقفاً، ثم نشغله:

```bash
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
```

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

```text
  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`:

```text
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` برسالة واضحة هذه المرة، بعد اختصار تفاصيل الموقع:

```text
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` ونعيد تشغيل المهمة:

```bash
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 سوف تكون كما يلي:

```text
op=r key={"id":2} before=null after={"id": 2, "name": "Omar", "email": "omar@example.com", "region": "Khobar", "phone": null} lsn=1563853864
op=r key={"id":3} before=null after={"id": 3, "name": "Lina", "email": "lina@example.com", "region": "Taif", "phone": null} lsn=1563853864
op=r key={"id":4} before=null after={"id": 4, "name": "Huda", "email": "huda@example.com", "region": "Makkah", "phone": null} lsn=1563853864
op=r key={"id":1} before=null after={"id": 1, "name": "Sara", "email": "sara@example.com", "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 القديمة بعد مدة أطول من أطول توقف تتوقعه، لا بعد ثوان.

### حذف المختبر

```bash
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 رسالة مضللة، والاستعادة بلقطة جديدة أعادت الحالة الأخيرة فقط.

[![](https://arabroot.io/content/images/2026/10/saga-outbox-cover-cover.webp)دليلالمعاملات الموزعة: Saga ونمط Outbox بدل 2PCكيف تنفذ عملية تمتد على عدة خدمات دون 2PC، بخطوات محلية وتعويضات وجدول Outbox لا يضيع فيه أي حدث.](https://arabroot.io/articles/%D8%A7%D9%84%D9%85%D8%B9%D8%A7%D9%85%D9%84%D8%A7%D8%AA-%D8%A7%D9%84%D9%85%D9%88%D8%B2%D8%B9%D8%A9-saga-%D9%88%D9%86%D9%85%D8%B7-outbox/)[![](https://arabroot.io/content/images/2026/10/kafka-cover-cover.webp)دليلApache Kafka: بث الأحداث بأحجام ضخمةكيف يحفظ عنقود Kafka أحداثك بنسخ على عدة سيرفرات، وماذا يحدث للمنتجين والمستهلكين عندما يسقط أحدها.](https://arabroot.io/articles/apache-kafka-%D8%A8%D8%AB-%D8%A7%D9%84%D8%A7%D8%AD%D8%AF%D8%A7%D8%AB-%D8%A8%D8%A7%D8%AD%D8%AC%D8%A7%D9%85-%D8%B6%D8%AE%D9%85%D8%A9/)[![](https://arabroot.io/content/images/2026/10/audit-trail-cover-cover.webp)دليلAudit Trail: كيف تبني سجل تدقيق لا يقبل العبثكيف تصمم سجلاً يجيب عن سؤال من غير هذه الصلاحية ومتى، وتحميه من الحذف والتعديل بعد كتابته.](https://arabroot.io/articles/%D8%B3%D8%AC%D9%84-%D8%A7%D9%84%D8%AA%D8%AF%D9%82%D9%8A%D9%82-audit-trail-%D9%85%D9%86-%D9%81%D8%B9%D9%84-%D9%85%D8%A7%D8%B0%D8%A7-%D9%88%D9%85%D8%AA%D9%89/)[![](https://arabroot.io/content/images/2026/10/queue-compare-cover-cover.webp)دليلRabbitMQ أم Kafka أم NATS؟ كيف تختار وسيط الرسائلكيف تختار بين RabbitMQ وKafka وNATS بحسب نموذج كل منها وضماناته وتكلفة تشغيله، مع قياس على جهاز واحد.](https://arabroot.io/articles/rabbitmq-%D8%A7%D9%85-kafka-%D8%A7%D9%85-nats-%D9%83%D9%8A%D9%81-%D8%AA%D8%AE%D8%AA%D8%A7%D8%B1-%D9%88%D8%B3%D9%8A%D8%B7-%D8%A7%D9%84%D8%B1%D8%B3%D8%A7%D8%A6%D9%84/)

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

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