Глоссарий — Debezium CDC Mastery

-- Просмотр текущей позиции WAL
SELECT pg_current_wal_lsn();
-- Результат: 0/16B9F08

-- Размер несинхронизированного WAL
SELECT pg_wal_lsn_diff(
  pg_current_wal_lsn(),
  confirmed_flush_lsn
) FROM pg_replication_slots
WHERE slot_name = 'debezium';

-- Текущий LSN
SELECT pg_current_wal_lsn();
-- Результат: 0/16B9F08

-- LSN в числовом формате (для расчёта лага)
SELECT pg_wal_lsn_diff('0/16B9F08', '0/16B9E00');
-- Результат: 264 байта

-- Просмотр активных слотов
SELECT slot_name, active, restart_lsn,
       confirmed_flush_lsn
FROM pg_replication_slots;

-- Создание слота для Debezium
SELECT pg_create_logical_replication_slot(
  'debezium',
  'pgoutput'
);

-- Удаление слота (потеря позиции!)
SELECT pg_drop_replication_slot('debezium');

-- Проверка доступных плагинов
SELECT * FROM pg_available_extensions
WHERE name = 'pgoutput';

-- Создание слота с плагином декодирования
SELECT pg_create_logical_replication_slot(
  'test_slot',
  'pgoutput'
);

-- Чтение изменений (для тестирования)
SELECT * FROM pg_logical_slot_peek_changes(
  'test_slot',
  NULL,
  NULL
);

-- Создание publication для всех таблиц
CREATE PUBLICATION dbz_publication
FOR ALL TABLES;

-- Publication для конкретных таблиц
CREATE PUBLICATION dbz_publication
FOR TABLE customers, orders;

-- Только INSERT и UPDATE (без DELETE)
CREATE PUBLICATION dbz_publication
FOR TABLE customers
WITH (publish = 'insert, update');

-- Просмотр publications
SELECT * FROM pg_publication;

-- PostgreSQL config для pgoutput
wal_level = logical
max_replication_slots = 4
max_wal_senders = 4

-- Connector config
{
  "plugin.name": "pgoutput",
  "publication.name": "dbz_publication"
}

-- Проверка использования pgoutput
SELECT slot_name, plugin
FROM pg_replication_slots;
-- Результат: debezium | pgoutput

-- Проверка включения binlog
SHOW VARIABLES LIKE 'log_bin';

-- Текущая позиция binlog
SHOW MASTER STATUS;

-- Список binlog файлов
SHOW BINARY LOGS;

-- Установка retention периода
SET GLOBAL expire_logs_days = 7;
SET GLOBAL binlog_expire_logs_seconds = 604800;

-- Проверка включения GTID
SHOW VARIABLES LIKE 'gtid_mode';

-- Текущий executed GTID set
SHOW MASTER STATUS;

-- GTID purged (удаленные binlog)
SHOW GLOBAL VARIABLES LIKE 'gtid_purged';

-- Просмотр текущего server_id
SHOW VARIABLES LIKE 'server_id';

-- my.cnf конфигурация
[mysqld]
server-id = 1
log-bin = mysql-bin
binlog_format = ROW

-- Connector config
{
  "database.server.id": "184054",
  "database.hostname": "mysql",
  "database.port": "3306"
}

-- Проверка формата binlog
SHOW VARIABLES LIKE 'binlog_format';

-- my.cnf конфигурация (обязательно для Debezium)
[mysqld]
binlog_format = ROW
binlog_row_image = FULL

-- Изменение в runtime (требует SUPER привилегии)
SET GLOBAL binlog_format = 'ROW';

-- Проверка binlog_row_image
SHOW VARIABLES LIKE 'binlog_row_image';

-- my.cnf конфигурация
[mysqld]
binlog_row_image = FULL

-- MINIMAL (экономит место, но Debezium не увидит неизменённые поля)
binlog_row_image = MINIMAL

-- Пример события с FULL
UPDATE users SET email = '[email protected]' WHERE id = 1;
-- Binlog содержит: id, name, email (до и после)

# Список топиков Debezium
kafka-topics --list --bootstrap-server localhost:9092 \
  | grep dbserver1

# Информация о топике
kafka-topics --describe \
  --topic dbserver1.inventory.customers \
  --bootstrap-server localhost:9092

# Чтение сообщений из топика
kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic dbserver1.inventory.customers \
  --from-beginning

# Просмотр текущих offset consumer group
kafka-consumer-groups --describe \
  --group connect-cluster \
  --bootstrap-server localhost:9092

# Сброс offset для повторного чтения
kafka-consumer-groups --reset-offsets \
  --group connect-cluster \
  --topic dbserver1.inventory.customers \
  --to-earliest --execute

# Создание топика с 3 партициями
kafka-topics --create \
  --topic dbserver1.inventory.orders \
  --partitions 3 \
  --replication-factor 2 \
  --bootstrap-server localhost:9092

# Просмотр партиций топика
kafka-topics --describe \
  --topic dbserver1.inventory.orders \
  --bootstrap-server localhost:9092

# Connector config - партиционирование по user_id
{
  "transforms": "partitionByUser",
  "transforms.partitionByUser.type":
    "org.apache.kafka.connect.transforms.ValueToKey",
  "transforms.partitionByUser.fields": "user_id"
}

# Список consumer groups
kafka-consumer-groups --list \
  --bootstrap-server localhost:9092

# Детали группы (offset, lag)
kafka-consumer-groups --describe \
  --group connect-debezium-postgres \
  --bootstrap-server localhost:9092

# Результат:
# GROUP       TOPIC         PARTITION  OFFSET  LAG
# connect-... customers     0          1523    0
# connect-... customers     1          1498    0
# connect-... orders        0          5832    12

// Connector config - snapshot mode
{
  "snapshot.mode": "initial",
  // Полный snapshot при первом запуске

  "snapshot.mode": "never",
  // Пропустить snapshot, только streaming

  "snapshot.mode": "schema_only",
  // Только структура таблиц, данные не копируются

  "snapshot.mode": "when_needed"
  // Snapshot если offset не найден
}

// Connector status - streaming phase
{
  "name": "inventory-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "kafka-connect:8083"
  },
  "tasks": [{
    "id": 0,
    "state": "RUNNING",
    "worker_id": "kafka-connect:8083"
  }],
  "type": "source"
}

# Создание коннектора через REST API
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @postgres-connector.json

# Проверка статуса
curl http://localhost:8083/connectors/inventory-connector/status

# Удаление коннектора
curl -X DELETE \
  http://localhost:8083/connectors/inventory-connector

// Connector config - фильтрация полей
{
  "transforms": "unwrap,addPrefix",

  // Debezium Unwrap - извлечь только 'after'
  "transforms.unwrap.type":
    "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.drop.tombstones": false,

  // Добавить префикс к топику
  "transforms.addPrefix.type":
    "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.addPrefix.regex": "(.*)",
  "transforms.addPrefix.replacement": "cdc.$1"
}

-- Outbox таблица
CREATE TABLE outbox (
  id UUID PRIMARY KEY,
  aggregatetype VARCHAR(255) NOT NULL,
  aggregateid VARCHAR(255) NOT NULL,
  type VARCHAR(255) NOT NULL,
  payload JSONB NOT NULL
);

-- Бизнес-логика (атомарная транзакция)
BEGIN;
  UPDATE orders SET status = 'PAID' WHERE id = 123;
  INSERT INTO outbox (id, aggregatetype, aggregateid, type, payload)
  VALUES (gen_random_uuid(), 'Order', '123', 'OrderPaid',
          '{"orderId": 123, "amount": 99.99}'::jsonb);
COMMIT;

// Debezium Outbox SMT
{
  "transforms": "outbox",
  "transforms.outbox.type":
    "io.debezium.transforms.outbox.EventRouter"
}

// Connector config - heartbeat каждые 5 минут
{
  "heartbeat.interval.ms": 300000,
  "heartbeat.topics.prefix": "__debezium-heartbeat",

  // Heartbeat событие
  {
    "ts_ms": 1686500577691,
    "source": {
      "connector": "postgresql",
      "name": "dbserver1",
      "lsn": 33023704
    }
  }
}

// Проверка heartbeat топика
kafka-console-consumer \
  --topic __debezium-heartbeat.dbserver1 \
  --bootstrap-server localhost:9092

// DELETE события от Debezium

// 1. DELETE событие (op='d')
{
  "before": {"id": 123, "name": "John"},
  "after": null,
  "op": "d"
}

// 2. Tombstone (следует сразу после)
{
  "key": {"id": 123},
  "value": null  // <- tombstone
}

// Connector config - управление tombstones
{
  "transforms": "unwrap",
  "transforms.unwrap.type":
    "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.drop.tombstones": false
}

// CDC Event Structure (Debezium JSON)
{
  "before": null,  // Состояние до изменения
  "after": {       // Новое состояние
    "id": 1001,
    "name": "John Doe",
    "email": "[email protected]"
  },
  "source": {
    "version": "2.1.0.Final",
    "connector": "postgresql",
    "name": "dbserver1",
    "ts_ms": 1486500577691,
    "db": "inventory",
    "schema": "public",
    "table": "customers",
    "lsn": 33023704
  },
  "op": "c",  // c=create, u=update, d=delete, r=read
  "ts_ms": 1486500584591
}

# Просмотр зарегистрированных схем
curl http://localhost:8081/subjects

# Получение схемы для топика
curl http://localhost:8081/subjects/\
dbserver1.inventory.customers-value/versions/latest

# Connector config с Schema Registry
{
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url":
    "http://schema-registry:8081"
}

// НЕ идемпотентная операция
UPDATE accounts SET balance = balance + 100
WHERE id = 123;
// Повторное применение = двойное начисление!

// Идемпотентная операция (event sourcing)
INSERT INTO account_events (account_id, type, amount, event_id)
VALUES (123, 'CREDIT', 100, 'evt-456')
ON CONFLICT (event_id) DO NOTHING;
// Повторное применение = игнорируется

// Debezium + Kafka idempotent producer
{
  "producer.enable.idempotence": true,
  "producer.max.in.flight.requests.per.connection": 5
}

// Kafka Streams - exactly-once
Properties props = new Properties();
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
  StreamsConfig.EXACTLY_ONCE_V2);

// Debezium config - основа для exactly-once
{
  // Kafka producer идемпотентность
  "producer.enable.idempotence": true,

  // Consumer должен сам обеспечить идемпотентность
  // через deduplication или event_id tracking
}

// Consumer pattern - deduplication
SET processed_events = new Set();
for (event in stream) {
  if (!processed_events.has(event.id)) {
    process(event);
    processed_events.add(event.id);
  }
}

// Event Store структура
CREATE TABLE event_store (
  event_id UUID PRIMARY KEY,
  aggregate_id VARCHAR(255),
  aggregate_type VARCHAR(255),
  event_type VARCHAR(255),
  event_data JSONB,
  created_at TIMESTAMP DEFAULT NOW()
);

// CDC события как Event Source
{
  "event_id": "evt-123",
  "aggregate_id": "order-456",
  "aggregate_type": "Order",
  "event_type": "OrderPlaced",
  "event_data": {
    "orderId": 456,
    "customerId": 789,
    "amount": 99.99
  }
}

// Восстановление состояния
SELECT * FROM event_store
WHERE aggregate_id = 'order-456'
ORDER BY created_at;