Знакомство с Debezium — CDC для Apache Kafka

Что за Debezium и вообще CDC?

Об архитектуре Debezium

• удаление неактуальных данных из кэша;
• отправка уведомлений;
• обновления поисковых индексов;
• некое подобие логов аудита;
• …

Конфигурация коннектора

1. источник данных, которым может являться MySQL начиная с версии 5.7, PostgreSQL 9.6+, MongoDB 3.2+ (полный список);
2. кластер Apache Kafka;
3. инстанс Kafka Connect (версии 1.x, 2.x);
4. сконфигурированный коннектор Debezium.

0. Kafka Connect

1. Настройка фреймворка Kafka Connect

• параметры подключения к кластеру,
• названия топиков, в которых будет храниться непосредственно конфигурация самого коннектора,
• имя группы, в которой запущен коннектор (в случае использования distributed-режима).

docker pull debezium/connect

• BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 — начальный список серверов кластера Kafka для получения полного списка членов кластера;
• OFFSET_STORAGE_TOPIC=connector-offsets — топик для хранения позиций, на которых на данный момент находится коннектор;
• CONNECT_STATUS_STORAGE_TOPIC=connector-status — топик для хранения статуса коннектора и его заданий;
• CONFIG_STORAGE_TOPIC=connector-config — топик для хранения данных конфигурации коннектора и его заданий;
• GROUP_ID=1 — идентификатор группы воркеров, на которых может выполняться задание коннектора; необходим при использовании распределённого (distributed) режима.

docker run \
  -e BOOTSTRAP_SERVERS='kafka-1:9092,kafka-2:9092,kafka-3:9092' \
  -e GROUP_ID=1 \
  -e CONFIG_STORAGE_TOPIC=my_connect_configs \
  -e OFFSET_STORAGE_TOPIC=my_connect_offsets \
  -e STATUS_STORAGE_TOPIC=my_connect_statuses  debezium/connect:1.2

Примечание про Avro

name: CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: VALUE_CONVERTER   
value: io.confluent.connect.avro.AvroConverter

2. Настройка самого коннектора

2.1. PostgreSQL

{
  "name": "pg-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "127.0.0.1",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "definitelynotpassword",
    "database.dbname" : "dbname",
    "database.server.name": "pg-dev",
    "table.include.list": "public.(.*)",
    "heartbeat.interval.ms": "5000",
    "slot.name": "dbname_debezium",
    "publication.name": "dbname_publication",
    "transforms": "AddPrefix",
    "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.AddPrefix.regex": "pg-dev.public.(.*)",
    "transforms.AddPrefix.replacement": "data.cdc.dbname"
  }
}

• При первом запуске он подключается к базе, указанной в конфигурации, и запускается в режиме initial snapshot, отправляя в Kafka начальный набор данных, полученных с помощью условного SELECT * FROM table_name.
• После того, как инициализация будет завершена, коннектор переходит в режим чтения изменений из WAL-файлов PostgreSQL.

• name — имя коннектора, для которого используется конфигурация, описанная ниже; в дальнейшем это имя используется для работы с коннектором (т.е. просмотра статуса/перезапуска/обновления конфигурации) через REST API Kafka Connect;
• connector.class — класс коннектора СУБД, который будет использоваться конфигурируемым коннектором;
• plugin.name — название плагина для логического декодирования данных из WAL-файлов. На выбор доступны wal2json, decoderbuffs и pgoutput. Первые два требуют установки соответствующих расширений в СУБД, а pgoutput для PostgreSQL версии 10 и выше не требует дополнительных манипуляций;
• database.* — опции для подключения к БД, где database.server.name — имя инстанса PostgreSQL, используемое для формирования имени топика в кластере Kafka;
• table.include.list — список таблиц, в которых мы хотим отслеживать изменения; задаётся в формате schema.table_name; нельзя использовать вместе с table.exclude.list;
• heartbeat.interval.ms — интервал (в миллисекундах), с которым коннектор отправляет heartbeat-сообщения в специальный топик;
• heartbeat.action.query — запрос, который будет выполняться при отправке каждого heartbeat-сообщения (опция появилась с версии 1.1);
• slot.name — имя слота репликации, который будет использоваться коннектором;
• publication.name — имя публикации в PostgreSQL, которую использует коннектор. В случае, если её не существует, Debezium попытается её создать. В случае, если у пользователя, под которым происходит подключение, недостаточно прав для этого действия — коннектор завершит работу с ошибкой;
• transforms определяет, как именно изменять название целевого топика:
  
  • transforms.AddPrefix.type указывает, что будем использовать регулярные выражения;
  • transforms.AddPrefix.regex — маска, по которой переопределяется название целевого топика;
  • transforms.AddPrefix.replacement — непосредственно то, на что переопределяем.

Подробнее про heartbeat и transforms

• Коннектор будет читать WAL-файлы и не обнаруживать в них коммита транзакций в те таблицы, за которыми он следит.
• Поэтому он не будет обновлять свою текущую позицию ни в топике, ни в слоте репликации.
• Это, в свою очередь, приведёт к «удержанию» WAL-файлов на диске и вероятному исчерпанию всего дискового пространства.

Ограничения коннектора

1. Функционал коннектора для PostgreSQL полагается на концепцию логического декодирования. Поэтому он не отслеживает запросы на изменение структуры БД (DDL) — соответственно, в топиках этих данных не будет.
2. Так как используются слоты репликации, подключение коннектора возможно только к ведущему экземпляру СУБД.
3. Если пользователю, под которым коннектор подключается к базе данных, выданы права только на чтение, то перед первым запуском потребуется вручную создать слот репликации и публикацию в БД.

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

curl -i -X POST -H "Accept:application/json" \
  -H  "Content-Type:application/json" \ http://localhost:8083/connectors/ \
  -d @pg-con.json

$ curl -i http://localhost:8083/connectors/pg-connector/status 
HTTP/1.1 200 OK
Date: Thu, 17 Sep 2020 20:19:40 GMT
Content-Type: application/json
Content-Length: 175
Server: Jetty(9.4.20.v20190813)

{"name":"pg-connector","connector":{"state":"RUNNING","worker_id":"172.24.0.5:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.24.0.5:8083"}],"type":"source"}

$ kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 \
  --from-beginning \
  --property print.key=true \
  --topic data.cdc.dbname

postgres=# insert into customers (id, first_name, last_name, email) values (1005, 'foo', 'bar', 'foo@bar.com');
INSERT 0 1
postgres=# update customers set first_name = 'egg' where id = 1005;
UPDATE 1

{
  "schema":{
    "type":"struct",
    "fields":[
      {
        "type":"int32",
        "optional":false,
        "field":"id"
      }
    ],
    "optional":false,
    "name":"data.cdc.dbname.Key"
  },
  "payload":{
    "id":1005
  }
}{
  "schema":{
    "type":"struct",
    "fields":[
      {
        "type":"struct",
        "fields":[
          {
            "type":"int32",
            "optional":false,
            "field":"id"
          },
          {
            "type":"string",
            "optional":false,
            "field":"first_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"last_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"email"
          }
        ],
        "optional":true,
        "name":"data.cdc.dbname.Value",
        "field":"before"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"int32",
            "optional":false,
            "field":"id"
          },
          {
            "type":"string",
            "optional":false,
            "field":"first_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"last_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"email"
          }
        ],
        "optional":true,
        "name":"data.cdc.dbname.Value",
        "field":"after"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":false,
            "field":"version"
          },
          {
            "type":"string",
            "optional":false,
            "field":"connector"
          },
          {
            "type":"string",
            "optional":false,
            "field":"name"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"ts_ms"
          },
          {
            "type":"string",
            "optional":true,
            "name":"io.debezium.data.Enum",
            "version":1,
            "parameters":{
              "allowed":"true,last,false"
            },
            "default":"false",
            "field":"snapshot"
          },
          {
            "type":"string",
            "optional":false,
            "field":"db"
          },
          {
            "type":"string",
            "optional":false,
            "field":"schema"
          },
          {
            "type":"string",
            "optional":false,
            "field":"table"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"txId"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"lsn"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"xmin"
          }
        ],
        "optional":false,
        "name":"io.debezium.connector.postgresql.Source",
        "field":"source"
      },
      {
        "type":"string",
        "optional":false,
        "field":"op"
      },
      {
        "type":"int64",
        "optional":true,
        "field":"ts_ms"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":false,
            "field":"id"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"total_order"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"data_collection_order"
          }
        ],
        "optional":true,
        "field":"transaction"
      }
    ],
    "optional":false,
    "name":"data.cdc.dbname.Envelope"
  },
  "payload":{
    "before":null,
    "after":{
      "id":1005,
      "first_name":"foo",
      "last_name":"bar",
      "email":"foo@bar.com"
    },
    "source":{
      "version":"1.2.3.Final",
      "connector":"postgresql",
      "name":"dbserver1",
      "ts_ms":1600374991648,
      "snapshot":"false",
      "db":"postgres",
      "schema":"public",
      "table":"customers",
      "txId":602,
      "lsn":34088472,
      "xmin":null
    },
    "op":"c",
    "ts_ms":1600374991762,
    "transaction":null
  }
}{
  "schema":{
    "type":"struct",
    "fields":[
      {
        "type":"int32",
        "optional":false,
        "field":"id"
      }
    ],
    "optional":false,
    "name":"data.cdc.dbname.Key"
  },
  "payload":{
    "id":1005
  }
}{
  "schema":{
    "type":"struct",
    "fields":[
      {
        "type":"struct",
        "fields":[
          {
            "type":"int32",
            "optional":false,
            "field":"id"
          },
          {
            "type":"string",
            "optional":false,
            "field":"first_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"last_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"email"
          }
        ],
        "optional":true,
        "name":"data.cdc.dbname.Value",
        "field":"before"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"int32",
            "optional":false,
            "field":"id"
          },
          {
            "type":"string",
            "optional":false,
            "field":"first_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"last_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"email"
          }
        ],
        "optional":true,
        "name":"data.cdc.dbname.Value",
        "field":"after"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":false,
            "field":"version"
          },
          {
            "type":"string",
            "optional":false,
            "field":"connector"
          },
          {
            "type":"string",
            "optional":false,
            "field":"name"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"ts_ms"
          },
          {
            "type":"string",
            "optional":true,
            "name":"io.debezium.data.Enum",
            "version":1,
            "parameters":{
              "allowed":"true,last,false"
            },
            "default":"false",
            "field":"snapshot"
          },
          {
            "type":"string",
            "optional":false,
            "field":"db"
          },
          {
            "type":"string",
            "optional":false,
            "field":"schema"
          },
          {
            "type":"string",
            "optional":false,
            "field":"table"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"txId"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"lsn"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"xmin"
          }
        ],
        "optional":false,
        "name":"io.debezium.connector.postgresql.Source",
        "field":"source"
      },
      {
        "type":"string",
        "optional":false,
        "field":"op"
      },
      {
        "type":"int64",
        "optional":true,
        "field":"ts_ms"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":false,
            "field":"id"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"total_order"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"data_collection_order"
          }
        ],
        "optional":true,
        "field":"transaction"
      }
    ],
    "optional":false,
    "name":"data.cdc.dbname.Envelope"
  },
  "payload":{
    "before":{
      "id":1005,
      "first_name":"foo",
      "last_name":"bar",
      "email":"foo@bar.com"
    },
    "after":{
      "id":1005,
      "first_name":"egg",
      "last_name":"bar",
      "email":"foo@bar.com"
    },
    "source":{
      "version":"1.2.3.Final",
      "connector":"postgresql",
      "name":"dbserver1",
      "ts_ms":1600375609365,
      "snapshot":"false",
      "db":"postgres",
      "schema":"public",
      "table":"customers",
      "txId":603,
      "lsn":34089688,
      "xmin":null
    },
    "op":"u",
    "ts_ms":1600375609778,
    "transaction":null
  }
}

• В случае с INSERT: значение до (before) равно null, а после — строка, которая была вставлена.
• В случае с UPDATE: в payload.before отображается предыдущее состояние строки, а в payload.after — новое с сутью изменений.

2.2 MongoDB

{
  "name": "mp-k8s-mongo-connector",
  "config": {
        "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
        "tasks.max": "1",
        "mongodb.hosts": "MainRepSet/mongo:27017",
        "mongodb.name": "mongo",
        "mongodb.user": "debezium",
        "mongodb.password": "dbname",
        "database.whitelist": "db_1,db_2",
        "transforms": "AddPrefix",
        "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.AddPrefix.regex": "mongo.([a-zA-Z_0-9]*).([a-zA-Z_0-9]*)",
        "transforms.AddPrefix.replacement": "data.cdc.mongo_$1"
        }
  }

Отказоустойчивость

1. Отказ Kafka Connect. Если Connect настроен на работу в распределенном режиме, для этого необходимо нескольким воркерам задать одинаковый group.id. Тогда при отказе одного из них коннектор будет перезапущен на другом воркере и продолжит чтение с последней коммитнутой позиции в топике в Kafka.
2. Потеря связности с Kafka-кластером. Коннектор просто остановит чтение на позиции, которую не удалось отправить в Kafka, и будет периодически пытаться повторно отправить её, пока попытка не завершится успехом.
3. Недоступность источника данных. Коннектор будет производить попытки повторного подключения к источнику в соответствии с конфигурацией. По умолчанию это 16 попыток с использованием exponential backoff. После 16-й неудачной попытки таск будет помечен как failed и потребуется его ручной перезапуск через REST-интерфейс Kafka Connect.
   
   • В случае с PostgreSQL данные не пропадут, т.к. использование слотов репликации не даст удалить WAL-файлы, не прочитанные коннектором. В этом случае есть и обратная сторона медали: если на продолжительное время будет нарушена сетевая связность между коннектором и СУБД, есть вероятность, что место на диске закончится, а это может может привести к отказу СУБД целиком.
   • В случае с MySQL файлы бинлогов могут быть отротированы самой СУБД раньше, чем восстановится связность. Это приведёт к тому, что коннектор перейдёт в состояние failed, а для восстановления нормального функционирования потребуется повторный запуск в режиме initial snapshot для продолжения чтения из бинлогов.
   • Про MongoDB. Документация гласит: поведение коннектора в случае, если файлы журналов/oplog'а были удалены и коннектор не может продолжить чтение с той позиции, где остановился, одинаково для всех СУБД. Оно заключается в том, что коннектор перейдёт в состояние failed и потребует повторного запуска в режиме initial snapshot.
     
     Однако бывают исключения. Если коннектор продолжительное время находился в отключенном состоянии (или не мог достучаться до экземпляра MongoDB), а oplog за это время прошёл ротацию, то при восстановлении подключения коннектор невозмутимо продолжит читать данные с первой доступной позиции, из-за чего часть данных в Kafka не попадёт.

Заключение

• JDBC Connector Kafka Connect;
• Несколько решений только для MySQL:
  
  • MySQL Streamer;
  • Maxwell;
  • SpinalTap;
• Oracle GoldenGate, но это совсем другая «весовая категория».

P.S.

• «Вышла werf 2.0: новый движок развёртывания Nelm и 300+ релизов за четыре года»;
• «Техническая история Kubernetes: секреты создателя
  »;
• «Некоторые моменты switchover в Patroni. Практические истории из SRE-будней. Часть 8».