Change Data Capture с Debezium

Что такое CDC?

Change Data Capture (CDC) — это паттерн передачи изменений в базе данных в виде событий (insert, update, delete) в реальном времени. На выходе у вас появляются стримы изменений, которые можно потреблять в микросервисах, аналитике, очередях и системах кэширования.

В конце гида вы получите рабочий стек: PostgreSQL 15 (релиз 2024), Kafka 3.5 (2025), Debezium 2.2 (релиз 2025) и пример consumer на Python, готовый читать CDC-топики. Время выполнения: ~60–90 минут на машине с Docker и 4 ГБ ОЗУ.

Что вы изучите

• Образ и запуск Debezium Connect 2.2 (релиз 2025) в Docker Compose.
• Настройку Debezium connector для PostgreSQL 15 с wal2json/plugin.
• Пример consumer-приложения на Python для чтения CDC из Kafka.
• Мониторинг коннектора и основные ошибки с быстрыми исправлениями.
• Стратегии обработки schema changes и миграций.

Требования

• ОС: Ubuntu 22.04 или CentOS Stream 9 / macOS 13+
• Docker Engine 24.0+ (релиз 2025), Docker Compose v2 встроенный.
• Минимум 4 ГБ ОЗУ и 2 CPU выделенные для среды разработки; для продакшна — 8+ ГБ и 4+ CPU.
• PostgreSQL 15 (релиз 2024) с wal_level = logical и расширением pgoutput/wal2json.
• Свободные порты: 5432 (Postgres), 9092 (Kafka), 8083 (Debezium Connect REST), 2181/9093 если используете Zookeeper — в примере используется встроенный Kafka‑KRaft; при необходимости — укажите иные порты.

Шаг 1: установка Debezium

Цель шага — поднять единую среду: Kafka 3.5 (KRaft), Zookeeper не обязателен, и Debezium Connect 2.2. Образ Debezium Connect (2.2.0) занимает примерно 220–300 МБ распакованного пространства в контейнере; время развертывания на NVMe — ~20–40 секунд за контейнер.

# файл docker-compose.yml
version: "3.8"
services:
  zookeeper: # опционально, если нужен ZK
    image: bitnami/zookeeper:3.8
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
    ports:
      - "2181:2181"
  kafka:
    image: confluentinc/cp-kafka:7.4
    environment:
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_HEAP_OPTS: -Xmx1024m -Xms256m
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper
  connect:
    image: debezium/connect:2.2
    environment:
      - BOOTSTRAP_SERVERS=kafka:9092
      - GROUP_ID=1
      - CONFIG_STORAGE_TOPIC=my_connect_configs
      - OFFSET_STORAGE_TOPIC=my_connect_offsets
      - STATUS_STORAGE_TOPIC=my_connect_statuses
      - REST_ADVERTISED_HOST_NAME=connect
      - CONNECT_KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter
      - CONNECT_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter
      - CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE=false
      - CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false
    ports:
      - "8083:8083"
    depends_on:
      - kafka
  postgres:
    image: postgres:15
    environment:
      POSTGRES_PASSWORD: postgres
    ports:
      - "5432:5432"
    volumes:
      - ./pgdata:/var/lib/postgresql/data

Команда для запуска:

docker compose up -d

Ожидаемый вывод:

$ docker compose up -d
Creating network "project_default" with the default driver
Creating project_zookeeper_1 ... done
Creating project_kafka_1     ... done
Creating project_postgres_1  ... done
Creating project_connect_1   ... done

Типовая ошибка: порт 8083 уже занят. Сообщение:

ERROR: for connect  Cannot start service connect: driver failed programming external connectivity on endpoint ... Bind for 0.0.0.0:8083 failed: port is already allocated

Фикс: освободить порт или пробросить на другой: в файле docker-compose изменить строку "8083:8083" на "18083:8083", затем перезапустить командой docker compose down && docker compose up -d.

Шаг 2: connector для PostgreSQL

Задача шага — зарегистрировать PostgreSQL-connector в Debezium Connect через REST API. Перед этим PostgreSQL должен быть настроен: в postgresql.conf укажите wal_level = logical, max_replication_slots >= 1, max_wal_senders >= 1. Для Postgres 15 (2024) это обязательные параметры.

# пример настройки (обычно /var/lib/postgresql/data/postgresql.conf)
wal_level = logical
max_replication_slots = 4
max_wal_senders = 4
# перезапустите postgres: systemctl restart postgresql или docker restart

Создайте пользователя для репликации:

-- выполняется внутри контейнера или psql
CREATE ROLE debezium WITH REPLICATION LOGIN PASSWORD 'dbz';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;

Пример JSON конфигурации коннектора:

POST http://localhost:8083/connectors
{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.dbname": "postgres",
    "database.server.name": "dbserver1",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "dbz_publication",
    "snapshot.mode": "initial",
    "topic.prefix": "dbserver1"
  }
}

Отправка запроса:

curl -i -X POST -H "Accept: application/json" -H "Content-Type: application/json" \
  --data '@connector-postgres.json' http://localhost:8083/connectors

Ожидаемый успешный вывод (сокращённо):

HTTP/1.1 201 Created
Content-Type: application/json

{"name":"inventory-connector","config":{...}}

Типичная ошибка 1: 409 Conflict — connector уже существует.

HTTP/1.1 409 Conflict
{"error_code":409,"message":"Connector inventory-connector already exists\