Debezium (CDC из PostgreSQL в Kafka): устройство и типовые операции

Общий документ для алертов группы debezium-alerts. Ранбуки отдельных алертов ссылаются на разделы отсюда.

1. Устройство

Debezium читает журнал изменений (WAL) PostgreSQL через слот логической репликации и публикует каждое изменение строки сообщением в Kafka. Потребители читают топики и поддерживают у себя копии таблиц.

Компонент	Где
Kafka Connect с плагином Debezium	кластер `yc-sarex-stage`, namespace `debezium`, Deployment `debezium-connect`, REST на порту 8083 (только внутри кластера)
Чарт, образ, конфигурация коннектора	репозиторий `infra/debezium`, ветка `stage`; конфигурация коннектора в `.helm/values.yaml`, ключ `connector.config`
Коннектор	`debezium-iams`, одна задача (task 0)
Источник	Managed PostgreSQL `c9qa2coo5ukgcg93fldm` (pg-stage), база `sarex_db`, таблицы `public.resource`, `public.resource_type`, `public.base_baseuser`
Слот репликации	`debezium_iams`
Publication	`debezium_iams`: три таблицы и служебная `debezium.heartbeat`
Пользователь PostgreSQL	`debezium`, секрет `debezium-postgresql` в namespace `debezium`
Kafka	Managed Kafka `c9qo2dr244cohj4amaia`, пользователь `debezium`, секрет `debezium-kafka`
Топики данных	`cdc.iams.public.resource`, `cdc.iams.public.resource_type`, `cdc.iams.public.base_baseuser`
Служебные топики	`debezium.connect.configs`, `debezium.connect.offsets`, `debezium.connect.status`, `cdc.iams.signal`, `debezium-heartbeat.cdc.iams`
Пользователи, топики, namespace	контракт terraform: `environments/stage/services/debezium.yaml`
Автоперезапуск упавшей задачи	CronJob `debezium-iams-task-restart`, каждые 5 минут
Метрики Kafka Connect и Debezium	JMX-экспортер в поде, порт 9404; Service `debezium-connect-metrics`, VMServiceScrape `debezium-connect`
Алерты	`kube-prometheus-stack`, ветка `stage`, `.helm/templates/alerts/debeziumAlerts.yaml`
Дашборд	Grafana stage, «Debezium / CDC» (`/d/debezium-cdc`)

Слот хранит позицию, до которой Debezium подтвердил чтение. PostgreSQL не удаляет WAL новее этой позиции. Если Debezium не подтверждает позицию, WAL копится. На кластере задан лимит max_slot_wal_keep_size 10 ГБ: при его превышении PostgreSQL сбрасывает слот (состояние lost), чтобы не заполнить диск. После сброса продолжить с прежней позиции нельзя.

Три таблицы меняются редко, а база в целом пишет много. Чтобы слот двигался, Debezium каждые 10 секунд обновляет строку в debezium.heartbeat (heartbeat.action.query). Это изменение проходит через слот, и Debezium подтверждает позицию.

2. Базовые команды

Имя пода:

POD=$(kubectl --context yc-sarex-stage -n debezium get pods -l app=debezium-connect -o jsonpath='{.items[0].metadata.name}')

Состояние коннектора и задачи:

kubectl --context yc-sarex-stage -n debezium exec $POD -- curl -s localhost:8083/connectors/debezium-iams/status

Коннектор и задача сообщают состояние раздельно. connector.state: RUNNING при tasks[0].state: FAILED означает, что данные не идут. В поле trace задачи находится причина падения.

Лог без шума реестра типов:

kubectl --context yc-sarex-stage -n debezium logs $POD --since=30m | grep -v 'already mapped' | grep -E ' (ERROR|WARN) '

Состояние слота (выполняется под любым пользователем базы sarex_db):

SELECT slot_name, active, active_pid, wal_status,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn))         AS retained,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS unconfirmed,
       pg_size_pretty(safe_wal_size)                                              AS before_lost
FROM pg_replication_slots WHERE slot_name = 'debezium_iams';

SELECT now() - ts AS heartbeat_age FROM debezium.heartbeat WHERE id = 1;

В норме: active = true, wal_status = reserved, retained десятки мегабайт, heartbeat_age менее 30 секунд.

3. Перезапуск задачи

kubectl --context yc-sarex-stage -n debezium exec $POD -- curl -s -o /dev/null -w '%{http_code}\n' -X POST localhost:8083/connectors/debezium-iams/tasks/0/restart

Ответ 204. Задача продолжает с подтверждённой позиции слота, данные не теряются.

После перезапуска задачи или пода в логе несколько секунд возможна ошибка replication slot "debezium_iams" is active for PID ...: предыдущее соединение репликации ещё не закрыто на стороне PostgreSQL. Debezium повторяет попытку до 6 раз с интервалом 10 секунд. Если попытки исчерпаны и задача в FAILED, перезапустить задачу ещё раз.

4. Повторный снапшот

Нужен, когда в топиках должны заново появиться все строки таблицы: потребитель пересобирает копию, слот был пересоздан, в таблицу добавлена колонка через ADD COLUMN ... DEFAULT (такое изменение не порождает событий по существующим строкам).

Снапшот запускается сообщением в топик cdc.iams.signal. Ключ сообщения обязан совпадать с topic.prefix коннектора (cdc.iams), иначе сигнал игнорируется.

kubectl --context yc-sarex-stage -n debezium exec $POD -- bash -c '
umask 077; CFG=/tmp/client.properties
{ echo "security.protocol=$KAFKA_SECURITY_PROTOCOL"
  echo "sasl.mechanism=$KAFKA_SASL_MECHANISM"
  echo "sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=\"$KAFKA_USERNAME\" password=\"$KAFKA_PASSWORD\";"
  echo "ssl.truststore.type=PEM"
  echo "ssl.truststore.location=$KAFKA_CA_FILE"; } > $CFG
echo "cdc.iams|{\"type\":\"execute-snapshot\",\"data\":{\"data-collections\":[\"public.resource\",\"public.resource_type\",\"public.base_baseuser\"],\"type\":\"INCREMENTAL\"}}" \
  | /kafka/bin/kafka-console-producer.sh --bootstrap-server "$BOOTSTRAP_SERVERS" --producer.config $CFG \
      --topic cdc.iams.signal --property parse.key=true --property key.separator="|" 2>/dev/null
rm -f $CFG'

В data-collections перечисляются только нужные таблицы. Проверка по логу:

kubectl --context yc-sarex-stage -n debezium logs $POD --since=5m | grep -E "Requested 'INCREMENTAL' snapshot|incremental snapshotting of table"

Для каждой таблицы должна появиться строка incremental snapshotting of table '...' finished. Снапшотные события приходят в те же топики с op: r и source.snapshot: incremental. Работа потока изменений во время снапшота не прерывается.

5. Пересоздание слота

Применяется, когда слот в состоянии lost или удалён.

Процедура не отрепетирована. Шаги составлены по документации Kafka Connect и PostgreSQL и на stage целиком не выполнялись. При первом применении сверять результат каждого шага, расхождения внести в этот раздел.

Остановить коннектор и сбросить сохранённую позицию:

kubectl --context yc-sarex-stage -n debezium exec $POD -- curl -s -o /dev/null -w '%{http_code}\n' -X PUT localhost:8083/connectors/debezium-iams/stop
kubectl --context yc-sarex-stage -n debezium exec $POD -- curl -s -X DELETE localhost:8083/connectors/debezium-iams/offsets

Удалить слот, если он существует (под пользователем debezium):

SELECT pg_drop_replication_slot('debezium_iams');

Запустить коннектор. Debezium создаст слот заново с текущей позиции:

kubectl --context yc-sarex-stage -n debezium exec $POD -- curl -s -o /dev/null -w '%{http_code}\n' -X PUT localhost:8083/connectors/debezium-iams/resume

Убедиться, что задача в RUNNING, слот active = true, wal_status = reserved.
Выполнить повторный снапшот по всем трём таблицам (раздел 4).
Сообщить потребителям (команды proc и PDM): изменения за период между сбросом слота и снапшотом переданы не событиями, а снапшотом. Удаления строк за этот период в топики не попали, копии у потребителей нужно сверить.

6. Добавление таблицы в поток

Список таблиц Debezium читает при старте задачи. Добавление таблицы только в publication не приводит к захвату.

Завести топик cdc.iams.public.<таблица> в контракте terraform (environments/stage/services/debezium.yaml), применить.
Под владельцем таблицы (sarex):

ALTER PUBLICATION debezium_iams ADD TABLE public.<таблица>;
GRANT SELECT ON public.<таблица> TO debezium;

Добавить таблицу в table.include.list и column.include.list в .helm/values.yaml репозитория infra/debezium, задеплоить. Job регистрации обновит конфигурацию, задача перезапустится.
Выполнить снапшот новой таблицы (раздел 4).

7. SQL вне контракта terraform

Следующие объекты созданы вручную под пользователем sarex в базе sarex_db. При пересоздании пользователя debezium или базы их нужно создать заново.

CREATE PUBLICATION debezium_iams FOR TABLE public.resource, public.resource_type,
  public.base_baseuser (id, username, first_name, last_name, email, service_account_id);
GRANT SELECT ON public.resource, public.resource_type, public.base_baseuser TO debezium;

CREATE SCHEMA debezium;
CREATE TABLE debezium.heartbeat (id int PRIMARY KEY, ts timestamptz NOT NULL);
GRANT USAGE ON SCHEMA debezium TO debezium;
GRANT SELECT, INSERT, UPDATE ON debezium.heartbeat TO debezium;
ALTER PUBLICATION debezium_iams ADD TABLE debezium.heartbeat;

SELECT на debezium.heartbeat обязателен: без него INSERT ... ON CONFLICT DO UPDATE завершается ошибкой permission denied.

8. Известные особенности

Строка ERROR Postgres roles LOGIN and REPLICATION are not assigned to user: debezium при старте ложная. Debezium проверяет атрибут роли, в Managed PostgreSQL право на репликацию выдаётся ролью mdb_replication.
Предупреждения Type [...] is already mapped (несколько сотен при старте) не являются ошибкой.
Канал сигналов требует signal.kafka.poll.timeout.ms больше нуля. При значении по умолчанию (0) соединение с Kafka по SASL_SSL не устанавливается, сигналы не читаются, в логе только socket connection setup timeout. В чарте задано 3000.
SASL-параметры Kafka Connect задаются в entrypoint.sh образа, а не переменными CONNECT_*: штатный entrypoint выводит значения CONNECT_* в лог.