Плагин kafka_cdc_plugin

← Стриминг / CDC Firebird

Плагин kafka_cdc_plugin предназначен для регистрации событий изменения данных в таблицах базы данных Firebird и их сохранения их в Kafka. Этот процесс также известен как Change Data Capture, далее CDC.

Каждое событие представляет собой документ в формате JSON, который максимально приближен к формату событий debezium. В следующей версии планируется добавить возможность сериализовать события в формате AVRO.

Требования

Для работы плагина требуется библиотеки librdkafka и liblz4. Под Windows они входят в дистрибутив. В Linux вы должны установить её вручную.

Установка в Debian:

sudo apt install librdkafka1

sudo apt install liblz4-1

Файлы и папки fb_streaming + kafka_cdc_plugin на Windows

• fb_streaming.exe - служба fb_streaming;
• fbclient.dll - клиентская библиотека Firebird (версия 4.0 и выше);
• icudt63.dll, icuin63.dll, icuuc63.dll, icudt63l.dat - файлы библиотеки ICU. В некоторых версиях Windows уже установлена системная библиотека ICU, в таком случае эти файлы необязательны;
• fb_streaming.conf - файл конфигурации службы fb_streaming;
• fb_streaming.log - журнал службы fb_streaming (создаётся самостоятельно);
• zlib1.dll, zstd.dll, lz4.dll - библиотеки сжатия данных и трафика, необходимые плагину kafka_cdc_plugin;
• libssl-3-x64.dll, libcrypto-3-x64.dll - библиотеки шифрования OpenSSL для windows-x64, libssl-3.dll, libcrypto-3.dll для, необходимые плагину kafka_cdc_plugin;
• libssl-3.dll, libcrypto-3.dll - библиотеки шифрования OpenSSL для windows-x86, необходимые плагину kafka_cdc_plugin;
• doc - папка с файлами документации;
• stream_plugins - папка для размещения файлов библиотек плагинов;
• stream_plugins/kafka_cdc_plugin.dll - библиотека плагина kafka_cdc_plugin.

[!NOTE] Замечание: SASL поддерживается только в 64-разрядной сборке.

Как это работает

Служба (демон) fb_streaming проверяет содержимое директории, указанной для задачи в конфигурационном файле fb_streaming.conf, и если в этой директории присутствует ещё не обработанные файлы журнала репликации, то она анализирует их, и генерирует соответствующие события, которые обрабатываются указанным плагином. Последний номер обработанного сегмента репликации записывается в контрольный файл с именем <database guid>. Расположение этого файла можно указать в параметре конфигурации lockDir. Если этот параметр не указан, то по умолчанию контрольный файл <database guid> будет создан в директории, указанной в качестве директории для журналов архивов для задачи.

[!IMPORTANT] Важно: Файлы журналов репликации должны быть в архивном состоянии.

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

• контрольную информацию (GUID базы данных, размер и т.д.);
• номер последнего обработанного сегмента и позицию в нём;
• номер сегмента с самой старой активной транзакцией (транзакция может начаться в одном сегменте, а завершиться в другом);
• список всех активных транзакций в виде пар {tnxNumber, segmentNumber}, где segmentNumber номер сегмента в котором транзакция началась.

Если произошла внештатная ситуация и служба fb_streaming была остановлена, то при следующем старте она читает контрольный файл, и перечитывает все сегменты репликации, начиная с номера сегмента с самой старой активной транзакцией, и заканчивая номером последнего обработанного сегмента. В процессе чтения этих сегментов fb_streaming повторяет все события из списка активных транзакций, после чего переходит в штатный режим работы.

Файл сегмента репликации удаляется, если его номер меньше чем номер сегмента с самой старой активной транзакцией.

Плагин kafka_cdc_plugin генерирует только события изменения данных для заданных таблиц для каждой из операции INSERT, UPDATE и DELETE. Для каждой активной транзакции существует свой буфер, в котором накапливаются эти события, и они отправляются в Kafka только по завершению транзакции, после чего буфер очищается.

События могут попадать в Kafka в двух форматах:

• JSON без схемы (параметры key_converter_schemas_enable и value_converter_schemas_enable равный false);
• JSON со схемой (параметры key_converter_schemas_enable и value_converter_schemas_enable равный true).

В будущих версиях планируется также добавить поддержку сериализации событий в формате AVRO, а также поддержку реестра схем.

Наименование топиков

По умолчанию kafka_cdc_plugin записывает события изменения для всех операций INSERT, UPDATE и DELETE, в одну тему Apache Kafka. kafka_cdc_plugin использует для имён топиков события изменения данных схему описанную в параметре kafka_topic_cdc_event, равную $[topicPrefix].cdc.

Здесь $[topicPrefix] - это подстановка, которая заменятся префиксом для топиков. Этот префикс задаётся параметром kafka_topic_prefix, по умолчанию он равен fb_streaming. Таким образом, по умолчанию все события изменения данных записываются в топик fb_streaming.cdc (после применения подстановки).

Существуют и другие подстановки. Подстановка $[tableName] обозначает имя таблицы для которой генерируется событие. Таким образом, указав в конфигурации

kafka_topic_cdc_event = $[topicPrefix].cdc.$[tableName]

мы получаем для каждой таблицы свой топик событий изменения данных.

Примеры полученных топиков с событиями изменения данных:

fb_streaming.cdc.products
fb_streaming.cdc.products_on_hand
fb_streaming.cdc.customers
fb_streaming.cdc.orders

Имя топика для события изменения метаданных указывается в параметре kafka_topic_transaction. В этом параметре доступна только подстановка $[topicPrefix].

Метаданные транзакции

Плагин kafka_cdc_plugin может генерировать события, которые представляют границы транзакций и дополняют сообщения о событиях изменения данных. По умолчанию генерация событий границ транзакций отключено. Её можно включить установив параметр transaction_metadata_enable в значение true.

Плагин kafka_cdc_plugin генерирует события границ транзакций BEGIN и END для каждой транзакции. События границ транзакций содержат следующие поля:

• status - BEGIN или END;
• id - идентификатор транзакции;
• ts_ms - время события границы транзакции (событие BEGIN или END). Поскольку в журналах репликации данного времени нет, то записывается время когда событие обрабатывается плагином kafka_cdc_plugin;
• event_count (для события END) - общее количество событий, созданных транзакцией;
• data_collections (для события END) - массив пар элементов data_collection и event_count, указывающий количество событий, которые kafka_cdc_plugin генерирует для изменений, происходящих для коллекции данных.

Пример:

{
    "status": "BEGIN",
    "id": 901,
    "ts_ms": 1713099401529,
    "event_count": null,
    "data_collections": null
}

{
    "status": "END",
    "id": 901,
    "ts_ms": 1713099404605,
    "event_count": 2,
    "data_collections": [
        {
            "data_collection": "fbstreaming.cdc.CUSTOMERS",
            "event_count": 2
        }
    ]
}

Если это не переопределено с помощью параметра kafka_topic_transaction, то kafka_cdc_plugin отправляет события транзакции в тему $[topicPrefix].transaction.

Расширение события изменения данных

Когда метаданные транзакции включены (transaction_metadata_enable = true), Envelope сообщения с данными пополняется новым полем transaction. Это поле предоставляет информацию о каждом событии в виде совокупности полей:

• id - идентификатор транзакции;
• total_order - абсолютная позиция события среди всех событий, созданных транзакцией;
• data_collection_order - позиция события в коллекции, среди всех событий созданных транзакцией.

Пример:

{
    "before": null,
    "after": {
        "ID": 20,
        "FIRST_NAME": "Anne",
        "LAST_NAME": "Kretchmar",
        "EMAIL": "[email protected]"
    },
    "source": {
       ...
    },
    "op": "c",
    "ts_ms": 1713099401533,
    "transaction": {
        "id": 901,
        "total_order": 1,
        "data_collection_order": 1
 }
}

Data Change Events (События изменения данных)

Плагин kafka_cdc_plugin генерирует событие изменения данных для каждой операции INSERT, UPDATE и DELETE на уровне записи. Каждое событие содержит ключ и значение. Структура ключа и значения зависят от измененной таблицы.

Служба fb_streaming и плагин kafka_cdc_plugin созданы для непрерывных потоков сообщений о событиях. Однако структура этих событий может со временем меняться, и потребителям может быть сложно это обработать. Чтобы решить эту проблему, каждое событие содержит схему для своего содержимого или, если вы используете реестр схем, идентификатор схемы, который потребитель может использовать для получения схемы из реестра. Это делает каждое событие самоописываемым.

Следующий скелет JSON показывает четыре основные части события изменения. Поле схемы находится в событии изменения только тогда, когда вы настраиваете плагин kafka_cdc_plugin для её создания. Аналогично, ключ события и полезные данные события находятся в событии изменения, только если вы настроили плагин kafka_cdc_plugin для их создания. Если вы используете формат JSON и настроили kafka_cdc_plugin для создания всех четырех основных частей событий изменений, события изменений имеют следующую структуру:

{
  "schema": { // <1>
    ...
   },
  "payload": { // <2>
    ...
  }
}
{
  "schema": { // <3>
    ...
  },
  "payload": { // <4>
    ...
  }
}

Описание основных частей события изменения

Ключи события изменения

Для заданной таблицы ключ события изменения имеет структуру, содержащую поле для каждого столбца первичного ключа таблицы на момент создания события.

Рассмотрим таблицу CUSTOMERS, и пример ключа события изменения для этой таблицы.

CREATE TABLE CUSTOMERS (
  ID BIGINT GENERATED BY DEFAULT AS IDENTITY,
  FIRST_NAME VARCHAR(255) NOT NULL,
  LAST_NAME VARCHAR(255) NOT NULL,
  EMAIL VARCHAR(255) NOT NULL,
  CONSTRAINT PK_CUSTOMERS PRIMARY KEY(ID)
);

{
    "schema": { // <1>
        "type": "struct",
        "name": "fbstreaming.CUSTOMERS.Key", // <2>
        "optional": "false", // <3>
        "fields": [ // <4>
            {
                "type": "int64",
                "optional": false,
                "field": "ID"
            }
        ]
    },
    "payload": { // <5>
        "ID": 1
    }
}

Описание основных частей ключа события изменения

Значения событий изменения

Значение в событии изменения немного сложнее, чем ключ. Как и ключ, значение имеет раздел schema и раздел payload. Раздел schema содержит схему, описывающую структуру Envelope раздела payload, включая её вложенные поля. События изменения для операций, которые создают, обновляют или удаляют данные, имеют значение payload со структурой Envelope.

Рассмотрим ту же таблицу, которая использовалась для демонстрации примера значения события изменения:

CREATE TABLE CUSTOMERS (
  ID BIGINT GENERATED BY DEFAULT AS IDENTITY,
  FIRST_NAME VARCHAR(255) NOT NULL,
  LAST_NAME VARCHAR(255) NOT NULL,
  EMAIL VARCHAR(255) NOT NULL,
  CONSTRAINT PK_CUSTOMERS PRIMARY KEY(ID)
);

Часть значения события изменения для изменения в этой таблице описывается для следующих операций:

• create events
• update events
• delete events

Для демонстрации этих событий запустим следующий набор запросов:

insert into CUSTOMERS (FIRST_NAME, LAST_NAME, EMAIL)
values ('Anne', 'Kretchmar', '[email protected]');

commit;

update CUSTOMERS
set FIRST_NAME = 'Anne Marie';

commit;

delete from CUSTOMERS;

commit;

Create events

В следующем примере показана часть значения события изменения, которое генерирует fb_streaming для операции, создающей данные в таблице CUSTOMERS:

{
    "schema": { // <1>
        "type": "struct",
        "fields": [
            {
                "type": "struct",
                "fields": [
                    {
                        "type": "int64",
                        "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": "fbstreaming.CUSTOMERS.Value", // <2>
                "field": "before"
            },
            {
                "type": "struct",
                "fields": [
                    {
                        "type": "int64",
                        "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": "fbstreaming.CUSTOMERS.Value",
                "field": "after"
            },
            {
                "type": "struct",
                "fields": [
                    {
                        "type": "string",
                        "optional": false,
                        "field": "dbguid"
                    },
                    {
                        "type": "int64",
                        "optional": false,
                        "field": "sequence"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "filename"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "table"
                    },
                    {
                        "type": "int64",
                        "optional": false,
                        "field": "tnx"
                    },
                    {
                        "type": "int64",
                        "optional": false,
                        "field": "ts_ms"
                    }
                ],
                "optional": false,
                "name": "fbstreaming.Source", // <3>
                "field": "source"
            },
            {
                "type": "string",
                "optional": false,
                "field": "op"
            },
            {
                "type": "int64",
                "optional": true,
                "field": "ts_ms"
            }
        ],
        "optional": false,
        "name": "fbstreaming.CUSTOMERS.Envelope" // <4>
    },
    "payload": { // <5>
       "before": null, // <6>
       "after": { // <7>
           "ID": 1,
           "FIRST_NAME": "Anne",
           "LAST_NAME": "Kretchmar",
           "EMAIL": "[email protected]"
        },
        "source": { // <8>
           "dbguid": "{9D66A972-A8B9-42E0-8542-82D1DA5F1692}",
           "sequence": 1,
           "filename": "TEST.FDB.journal-000000001",
           "table": "CUSTOMERS",
           "tnx": 200,
           "ts_ms": 1711288254908
        },
        "op": "c", // <9>
        "ts_ms": 1711288255056 // <10>
    }
}

Описание основных частей события create:

Update events

Значение события изменения для операции update в примере таблицы CUSTOMERS имеет ту же схему, что и событие create для этой таблицы. Аналогично, полезная нагрузка значения события имеет ту же структуру. Однако полезная нагрузка значения события содержит разные значения в событии update. Вот пример значения события изменения в событии, которое fb_streaming генерирует для обновления в таблице CUSTOMERS:

{
    "schema": { ... },
    "payload": {
        "before": { // <1>
            "ID": 1,
            "FIRST_NAME": "Anne",
            "LAST_NAME": "Kretchmar",
            "EMAIL": "[email protected]"
        },
        "after": { // <2>
            "ID": 1,
            "FIRST_NAME": "Anne Marie",
            "LAST_NAME": "Kretchmar",
            "EMAIL": "[email protected]"
        },
        "source": { // <3>
            "dbguid": "{9D66A972-A8B9-42E0-8542-82D1DA5F1692}",
            "sequence": 2,
            "filename": "TEST.FDB.journal-000000002",
            "table": "CUSTOMERS",
            "tnx": 219,
            "ts_ms": 1711288254908
        },
        "op": "u", // <4>
        "ts_ms": 1711288256121 // <5>
    }
}

Описание основных частей события update:

Delete events

Значение в событии изменения delete имеет ту же часть схемы, что и события create и update для той же таблицы. Часть payload в событии delete для примера таблицы CUSTOMERS выглядит следующим образом:

{
    "schema": { ... },
    "payload": { // <1>
        "before": {
            "ID": 1,
            "FIRST_NAME": "Anne Marie",
            "LAST_NAME": "Kretchmar",
            "EMAIL": "[email protected]"
        },
        "after": null, // <2>
        "source": { // <3>
            "dbguid": "{9D66A972-A8B9-42E0-8542-82D1DA5F1692}",
            "sequence": 3,
            "filename": "TEST.FDB.journal-000000003",
            "table": "CUSTOMERS",
            "tnx": 258,
            "ts_ms": 1711288254908
        },
        "op": "d", // <4>
        "ts_ms": 1711288256152 // <5>
    }
}

Описание основных частей события delete:

Отображение типов данных

Запуск Change Data Capture

Подробно опишем шаги необходимые для запуска сбора изменений (Change Data Capture) на вашей базе данных:

1. Настройка Kafka
2. Настройка Firebird и подготовка базы данных
3. Настройка службы fb_streaming и плагина kafka_cdc_plugin
4. Запуск Kafka
5. Установка и старт службы fb_streaming
6. Старт публикации в базе данных

Настройка Kafka

Для тестирования плагина kafka_cdc_plugin, используется настроенная установка Kafka в docker. Для этого используется docker-compose.yml со следующим содержимым:

version: "2"

services:

  zookeeper:
    image: confluentinc/cp-zookeeper:7.2.1
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-server:7.2.1
    hostname: kafka
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9997:9997"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9997
      KAFKA_JMX_HOSTNAME: kafka

  kafka-ui:
    container_name: kafka-ui
    image: provectuslabs/kafka-ui:latest
    ports:
      - 8080:8080
    environment:
      DYNAMIC_CONFIG_ENABLED: 'true'
    volumes:
      - "d:\\docker\\kafka\\config.yml:/etc/kafkaui/dynamic_config.yaml"

Подключаемый файл config.yml содержит:

auth:
  type: DISABLED
kafka:
  clusters:
  - bootstrapServers: kafka:29092
    name: Kafka CDC Cluster
    properties: {}
    readOnly: false
rbac:
  roles: []
webclient: {}

Настройка Firebird и подготовка базы данных

Теперь необходимо настроить асинхронную репликацию для вашей базы данных, для этого в файле replication.conf необходимо добавить следующие строчки:

database = c:\fbdata\5.0\test.fdb
{
   journal_directory = d:\fbdata\5.0\replication\test\journal
   journal_archive_directory = d:\fbdata\5.0\replication\test\archive
   journal_archive_command = "copy $(pathname) $(archivepathname) && copy $(pathname) d:\fbdata\5.0\replication\test\kafka_source"
   journal_archive_timeout = 10
}

Обратите внимание: здесь происходит дублирование файлов архивов журналов, чтобы одновременно работала логическая репликация и задача по отправки событий в Kafka. Это необходимо, поскольку файлы с архивами журналов удаляются после обработки и не могут быть использованы другой задаче.

Если журналы репликации не используются для самой репликации, а только необходимы для Change Data Capture, то конфигурацию можно упростить:

database = c:\fbdata\5.0\test.fdb
{
   journal_directory = d:\fbdata\5.0\replication\test\journal
   journal_archive_directory = d:\fbdata\5.0\replication\test\kafka_source
   journal_archive_timeout = 10
}

Теперь надо включить необходимые таблицы в публикацию. Для примера выше достаточно добавить в публикацию таблицу CUSTOMERS. Это делается следующим запросом:

ALTER DATABASE INCLUDE CUSTOMERS TO PUBLICATION;

или можно включить в публикацию сразу все таблицы базы данных:

ALTER DATABASE INCLUDE ALL TO PUBLICATION;

Настройка службы fb_streaming и плагина kafka_cdc_plugin

Далее настроим конфигурацию fb_streaming.conf для того, чтобы fb_streaming автоматически отправлял события изменения данных в Kafka.

task = d:\fbdata\5.0\replication\test\kafka_source
{
    database = inet://localhost:3055/test
    username = SYSDBA
    password = masterkey
    deleteProcessedFile = true
    plugin = kafka_cdc_plugin
    dumpBlobs = true
    kafka_brokers = localhost:9092
    kafka_topic_prefix = fb_streaming
    kafka_topic_cdc_event = $[topicPrefix].cdc
    kafka_topic_transaction = $[topicPrefix].transaction
    key_cdc_events_enable = true
    key_converter_schemas_enable = true
    value_converter_schemas_enable = true
    transaction_metadata_enable = true
}

В Linux эта конфигурация будет выглядеть так:

task = /mnt/d/fbdata/5.0/replication/test/kafka_source
{
    database = inet://localhost:3055/test
    username = SYSDBA
    password = masterkey
    deleteProcessedFile = true
    plugin = kafka_cdc_plugin
    dumpBlobs = true
    kafka_brokers = localhost:9092
    kafka_topic_prefix = fb_streaming
    kafka_topic_cdc_event = $[topicPrefix].cdc
    kafka_topic_transaction = $[topicPrefix].transaction
    key_cdc_events_enable = true
    key_converter_schemas_enable = true
    value_converter_schemas_enable = true
    transaction_metadata_enable = true
}

Параметр task описывает задачу для выполнения службой fb_streaming. Он указывает папку, в которой расположены файлы сегментов репликации для их обработки плагином. Таких задач может быть несколько. Этот параметр является сложным и сам описывает конфигурацию конкретной задачи. Опишем параметры доступные для задачи выполняемой плагином kafka_cdc_plugin:

• controlFileDir - директория в которой будет создан контрольный файл (по умолчанию та же директория, что и sourceDir);
• database - строка подключения к базе данных (обязательный);
• username - имя пользователя для подключения к базе данных;
• password - пароль для подключения к базе данных;
• plugin - плагин, который обрабатывает события, возникающие в процессе анализа журнала репликации (обязательный);
• deleteProcessedFile - удалять ли файл журнала после обработки (по умолчанию true). Этот параметр полезно ставить в значение false для отладки, когда одни и те же журналы надо обработать многократно не удаляя их;
• warn_tnx_not_found - генерировать предупреждение вместо ошибки, если транзакция не найдена в сегментах репликации. Если этот параметр установлен в true, то в журнал fb_streaming будет записано соответствующее предупреждение, содержимое потерянной транзакции будет проигнорировано, и плагин продолжит свою работу. По умолчанию установлен в false;
• errorTimeout - тайм-аут после ошибки, в секундах. По истечению этого таймаута будет произведено повторное сканирование сегментов и перезапуск задачи. По умолчанию равен 60 секунд;
• include_tables - регулярное выражение, определяющие имена таблиц для которых необходимо отслеживать события;
• exclude_tables - регулярное выражение, определяющие имена таблиц для которых не надо отслеживать события;
• dumpBlobs - публиковать ли новые значения BLOB полей (по умолчанию false);
• charset_for_none_fields - это набор символов, который используется для преобразования данных из строковых полей с типом кодировки NONE. Обычно данные в строковых полях с набором символов NONE хранятся “как есть”, то есть в исходной кодировке, которую использовало приложение при сохранении. Firebird не перекодирует такие строки. Формат JSON требует, чтобы все строки были в кодировке UTF8. Чтобы корректно конвертировать строки из полей NONE в UTF8, системе необходимо знать исходную кодировку данных. Именно её и задает параметр charset_for_none_fields;
• kafka_brokers - адреса брокеров Kafka. Можно указать несколько адресов. Адреса разделяются запятой;
• kafka_topic_prefix - префикс топиков Kafka. Он задаёт макроподстановку $[topicPrefix], которая может использоваться для имён топиков.
• kafka_topic_cdc_event - имя топика(ов) Kafka, в которой сохраняются события изменения данных. Могут использоваться макроподстановки $[topicPrefix] и $[tableName];
• kafka_topic_transaction - имя топика Kafka, в которой сохраняются метаданные транзакций. Может использоваться макроподстановка $[topicPrefix];
• async_producer - если установить этот параметр в true, то сообщения в Kafka будут отправляться асинхронно, то есть fb_streaming не будет ждать подтверждения об успешной отправке сообщения, а сразу переходить к обработке следующего сообщения. Сообщения отправляются асинхронно до тех пор, пока внутренний буфер не отправленных сообщений не будет заполнен. По умолчанию размер этого буфера равен 500.
• key_cdc_events_enable - если эта опция установлена в true, то каждое событие изменения данных содержит информацию о ключе, в противном случае ключ события будет равен null. Это может быть полезно, поскольку Kafka использует ключ события для секционирования;
• key_converter_schemas_enable - включать ли в ключ события обновления данных схему;
• value_converter_schemas_enable - включать ли в значение события обновления данных схему;
• transaction_metadata_enable - отправлять ли события старта и завершения транзакции.

Далее описаны параметры библиотеки LibRdKafka, используемые этим плагином. В основном эти параметры связаны с безопасностью соединения с серверами Kafka. Каждый параметр имеет префикс kafka_. Точки внутри параметра заменяются подчеркиваниями _. Описание этих параметров для библиотеки LibRdKafka можно посмотреть по следующей ссылке LibRdKafka CONFIGURATION.

• kafka_security_protocol - Протокол, используемый для связи с брокерами. Допустимые значения: plaintext, ssl, sasl_plaintext, sasl_ssl. В библиотеке LibRdKafka этот параметр имеет имя security.protocol. Значение по умолчанию: plaintext.
• kafka_ssl_cipher_suites - Список наборов шифров. Это именованная комбинация аутентификации, шифрования, MAC и алгоритма обмена ключами, используемая для согласования настроек безопасности для сетевого соединения с использованием сетевого протокола TLS или SSL. По умолчанию поддерживаются все доступные наборы шифров. В библиотеке LibRdKafka этот параметр имеет имя ssl.cipher.suites.
• kafka_ssl_curves_list - Поддерживаемое curves расширение в сообщении TLS ClientHello указывает curves (стандартные/именованные или явные GF(2^k) или GF(p)), которые клиент готов использовать на сервере. См. страницу руководства для SSL_CTX_set1_curves_list(3). Требуется OpenSSL >= 1.0.2. В библиотеке LibRdKafka этот параметр имеет имя ssl.curves.list.
• kafka_ssl_sigalgs_list - Клиент использует расширение TLS ClientHello signature_algorithms, чтобы указать серверу, какие пары алгоритмов подписи/хэша могут использоваться в цифровых подписях. См. страницу руководства для SSL_CTX_set1_sigalgs_list(3). Требуется OpenSSL >= 1.0.2. В библиотеке LibRdKafka этот параметр имеет имя ssl.sigalgs.list.
• kafka_ssl_key_location - Путь к закрытому ключу клиента (PEM), используемому для аутентификации. В библиотеке LibRdKafka этот параметр имеет имя ssl.key.location.
• kafka_ssl_key_password - Парольная фраза закрытого ключа (для использования с ssl.key.location и set_ssl_cert()). В библиотеке LibRdKafka этот параметр имеет имя ssl.key.password.
• kafka_ssl_certificate_location - Путь к открытому ключу клиента (PEM), используемому для аутентификации. В библиотеке LibRdKafka этот параметр имеет имя ssl.certificate.location.
• kafka_ssl_ca_location - Путь к файлу или каталогу сертификата(ов) CA для проверки ключа брокера. Значения по умолчанию: в Windows системные сертификаты CA автоматически ищутся в корневом хранилище сертификатов Windows. В Mac OSX эта конфигурация по умолчанию использует проверку. Рекомендуется установить openssl с помощью Homebrew, чтобы предоставить сертификаты CA. В Linux установите пакет ca-certificates из дистрибутива. Если OpenSSL связан статически или ssl.ca.location настроен на проверку, будет проверен список стандартных путей, и первый найденный путь будет использован в качестве пути расположения сертификата CA по умолчанию. Если OpenSSL связан динамически, то будет использован путь по умолчанию библиотеки OpenSSL (см. OPENSSLDIR в openssl версии -a). В библиотеке LibRdKafka этот параметр имеет имя ssl.ca.location.
• kafka_ssl_ca_certificate_stores - Список хранилищ сертификатов Windows, разделенных запятыми, из которых следует загрузить сертификаты CA. Сертификаты будут загружены в том же порядке, в котором указаны хранилища. Если ни из одного из указанных хранилищ не удается загрузить сертификаты, регистрируется ошибка и вместо этого используется местоположение CA библиотеки OpenSSL по умолчанию. Хранилища обычно имеют одно или несколько из следующих имён: MY, Root, Trust, CA. В библиотеке LibRdKafka этот параметр имеет имя ssl.ca.certificate.stores. Значение по умолчанию: Root.
• kafka_ssl_crl_location - Путь к CRL для проверки действительности сертификата брокера. В библиотеке LibRdKafka этот параметр имеет имя ssl.crl.location.
• kafka_ssl_keystore_location - Путь к хранилищу ключей клиента (PKCS#12), используемому для аутентификации. В библиотеке LibRdKafka этот параметр имеет имя ssl.keystore.location.
• kafka_ssl_keystore_password - Пароль хранилища ключей клиента (PKCS#12). В библиотеке LibRdKafka этот параметр имеет имя ssl.keystore.password.
• kafka_ssl_providers - Список поставщиков реализации OpenSSL 3.0.x, разделенный запятыми. Например, default,legacy. В библиотеке LibRdKafka этот параметр имеет имя ssl.providers.
• kafka_ssl_engine_id - Идентификатор движка OpenSSL - это имя, используемое для загрузки движка. В библиотеке LibRdKafka этот параметр имеет имя ssl.engine.id. Значение по умолчанию: dynamic.
• kafka_enable_ssl_certificate_verification - Включить встроенную проверку сертификата брокера (сервера) OpenSSL. Допустимые значения: true, false. В библиотеке LibRdKafka этот параметр имеет имя kafka_enable_ssl_certificate_verification. Значение по умолчанию: true.
• kafka_ssl_endpoint_identification_algorithm - Алгоритм идентификации конечной точки для проверки имени хоста брокера с использованием сертификата брокера. Допустимые значения: none, https. https - проверка имени хоста сервера (брокера), как указано в RFC2818. none - проверка конечной точки не требуется. Требуется OpenSSL >= 1.0.2. В библиотеке LibRdKafka этот параметр имеет имя ssl.endpoint.identification.algorithm. Значение по умолчанию: https.
• kafka_sasl_mechanism - Механизм SASL для использования при аутентификации. Поддерживается: GSSAPI, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER. Замечание: несмотря на название, необходимо настроить только один механизм. В библиотеке LibRdKafka этот параметр имеет имя sasl.mechanism. Значение по умолчанию: GSSAPI.
• kafka_sasl_username - Имя пользователя SASL для использования с механизмами PLAIN и SASL-SCRAM-... В библиотеке LibRdKafka этот параметр имеет имя sasl.username.
• kafka_sasl_password - Пароль SASL для использования с механизмами PLAIN и SASL-SCRAM-... В библиотеке LibRdKafka этот параметр имеет имя sasl.password.
• kafka_sasl_kerberos_service_name - Основное имя Kerberos, под которым работает Kafka, не включая /hostname@REALM. В библиотеке LibRdKafka этот параметр имеет имя sasl.kerberos.service.name. Значение по умолчанию: kafka.
• kafka_sasl_kerberos_principal - Основное имя Kerberos этого клиента. (Не поддерживается в Windows, будет использоваться имя основное пользователя, вошедшего в систему). В библиотеке LibRdKafka этот параметр имеет имя sasl.kerberos.principal. Значение по умолчанию: kafkaclient.
• kafka_sasl_oauthbearer_config - Конфигурация SASL/OAUTHBEARER. Формат зависит от реализации и должен быть проанализирован соответствующим образом. Реализация незащищенного токена по умолчанию (см. https://tools.ietf.org/html/rfc7515#appendix-A.5) распознает разделенные пробелами пары name=value с допустимыми именами, включая principalClaimName, principal, scopeClaimName, scope и lifeSeconds. Значение по умолчанию для principalClaimName - sub, значение по умолчанию для scopeClaimName - scope, а значение по умолчанию для lifeSeconds - 3600. Значение scope - это формат CSV, при этом значение по умолчанию - no/empty scope. Например: principalClaimName=azp principal=admin scopeClaimName=roles scope=role1,role2 lifeSeconds=600. Кроме того, расширения SASL могут быть переданы брокеру через extension_NAME=value. Например: principal=admin extension_traceId=123. В библиотеке LibRdKafka этот параметр имеет имя sasl.oauthbearer.config.
• kafka_sasl_oauthbearer_method - Установите значение default или oidc, чтобы контролировать, какой метод входа будет использоваться. Если установлено значение oidc, необходимо также указать следующие свойства: kafka_sasl_oauthbearer_client_id, kafka_sasl_oauthbearer_client_secret и kafka_sasl_oauthbearer_token_endpoint_url (sasl.oauthbearer.client.id, sasl.oauthbearer.client.secret и sasl.oauthbearer.token.endpoint.url). В библиотеке LibRdKafka этот параметр имеет имя sasl.oauthbearer.method. Значение по умолчанию: default.
• kafka_sasl_oauthbearer_client_id - Открытый идентификатор приложения. Должен быть уникальным среди всех клиентов, которых обрабатывает сервер авторизации. Используется только в том случае, если kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) установлен в значение oidc. В библиотеке LibRdKafka этот параметр имеет имя sasl.oauthbearer.client.id.
• kafka_sasl_oauthbearer_client_secret - Секрет клиента известен только приложению и серверу авторизации. Это должна быть достаточно случайная строка, которую невозможно угадать. Используется только в том случае, если kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) установлен в значение oidc. В библиотеке LibRdKafka этот параметр имеет имя sasl.oauthbearer.client.secret.
• kafka_sasl_oauthbearer_scope - Клиент использует это для указания области запроса доступа к брокеру. Используется только когда kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) установлен в значение oidc. В библиотеке LibRdKafka этот параметр имеет имя sasl.oauthbearer.scope.
• kafka_sasl_oauthbearer_extensions - Разрешает предоставление брокеру дополнительной информации. Разделенный запятыми список пар key=value. Например, supportFeatureX=true,organizationId=sales-emea. Используется только в том случае, если kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) установлен в значение oidc. В библиотеке LibRdKafka этот параметр имеет имя sasl.oauthbearer.extensions.
• kafka_sasl_oauthbearer_token_endpoint_url - HTTP(S) URI конечной точки токена эмитента OAuth/OIDC, используемый для получения токена. Используется только когда kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) установлен в oidc. В библиотеке LibRdKafka этот параметр имеет имя sasl.oauthbearer.token.endpoint.url.

Запуск Kafka

Теперь можно запустить docker с контейнером Kafka:

docker-compose up -d

Для остановки docker выполните:

docker-compose down

Установка и старт службы fb_streaming

Следующим шагом необходимо установить и запустить службу fb_streaming.

В Windows это делается следующими командами (необходимы права Администратора):

fb_streaming install
fb_streaming start

В Linux:

sudo systemctl enable fb_streaming

sudo systemctl start fb_streaming

[!NOTE] Для тестирования работы fb_streaming без установки службы просто наберите команду fb_streaming без аргументов. fb_streaming будет запущен как приложение и завершён после нажатия клавиши Enter.

Старт публикации в базе данных

После того, как вы всё настроили и запустили, необходимо разрешить публикацию в вашей базе данных. Это делается следующим SQL запросом:

ALTER DATABASE ENABLE PUBLICATION;

С этого момента служба fb_streaming будет отслеживать изменения в указанных таблицах и публиковать в топике fb_streaming.cdc, на серверах указных в kafka_brokers.